How we tuned JVM 21 ZGC for Apache Flink 1.20 to process 500k events/sec with no GC pause

\nimport org.apache.flink.api.common.eventtime.WatermarkStrategy;\nimport org.apache.flink.api.common.serialization.SimpleStringSchema;\nimport org.apache.flink.connector.kafka.source.KafkaSource;\nimport org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;\nimport org.apache.flink.streaming.api.datastream.DataStream;\nimport org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;\nimport org.apache.flink.streaming.api.functions.sink.SinkFunction;\nimport org.apache.flink.util.Collector;\nimport org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;\nimport org.apache.flink.api.common.state.ValueState;\nimport org.apache.flink.api.common.state.ValueStateDescriptor;\nimport org.apache.flink.configuration.Configuration;\nimport org.apache.flink.streaming.api.functions.KeyedProcessFunction;\nimport org.apache.flink.streaming.api.CheckpointingMode;\n\nimport java.time.Duration;\nimport java.util.Properties;\n\n/**\n * Flink 1.20 event processing job tuned for JVM 21 ZGC compatibility.\n * Configures state backends, checkpointing, and source/sink to minimize GC pressure.\n */\npublic class ZGCTunedFlinkJob {\n    private static final String JOB_NAME = \"zgc-tuned-flink-1.20-event-processor\";\n    private static final int PARALLELISM = 8; // Matches number of vCPUs per task manager\n    private static final long CHECKPOINT_INTERVAL_MS = 5000L;\n\n    public static void main(String[] args) {\n        // 1. Initialize Flink execution environment with ZGC-aware configs\n        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();\n        env.setParallelism(PARALLELISM);\n        env.enableCheckpointing(CHECKPOINT_INTERVAL_MS, CheckpointingMode.EXACTLY_ONCE);\n\n        // Configure checkpoint storage to avoid RocksDB lock contention (reduces GC pressure)\n        env.getCheckpointConfig().setCheckpointStorage(\"file:///opt/flink/checkpoints\");\n        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000L);\n        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);\n\n        // 2. Configure Kafka source with bounded fetch to prevent memory spikes\n        Properties kafkaProps = new Properties();\n        kafkaProps.put(\"bootstrap.servers\", \"kafka-broker-01:9092,kafka-broker-02:9092\");\n        kafkaProps.put(\"group.id\", \"flink-zgc-consumer-group\");\n        kafkaProps.put(\"max.partition.fetch.bytes\", \"1048576\"); // 1MB per partition to limit heap usage\n\n        KafkaSource kafkaSource = KafkaSource.builder()\n                .setTopics(\"raw-events-topic\")\n                .setProperties(kafkaProps)\n                .setStartingOffsets(OffsetsInitializer.latest())\n                .setValueOnlyDeserializer(new SimpleStringSchema())\n                .build();\n\n        // 3. Add source with watermark strategy for event time processing\n        DataStream rawEvents = env.fromSource(\n                kafkaSource,\n                WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5))\n                        .withTimestampAssigner((event, timestamp) -> System.currentTimeMillis()),\n                \"Kafka Source\"\n        );\n\n        // 4. Process events with keyed state (optimized for ZGC concurrent compaction)\n        SingleOutputStreamOperator processedEvents = rawEvents\n                .keyBy(event -> event.split(\",\")[0]) // Key by user ID\n                .process(new ZGCStatefulProcessor());\n\n        // 5. Sink to downstream topic with error handling\n        processedEvents.addSink(new SinkFunction() {\n            @Override\n            public void invoke(String value, Context context) {\n                try {\n                    // Simulate downstream write; add retry logic for transient errors\n                    if (value == null || value.isEmpty()) {\n                        throw new IllegalArgumentException(\"Empty event received for sink\");\n                    }\n                    // In production, replace with Kafka/Redis sink\n                    System.out.println(\"Processed event: \" + value);\n                } catch (Exception e) {\n                    System.err.println(\"Sink failed for event: \" + e.getMessage());\n                    // Log to error topic instead of throwing to avoid job failure\n                }\n            }\n        });\n\n        // 6. Execute job with error handling for startup failures\n        try {\n            env.execute(JOB_NAME);\n        } catch (Exception e) {\n            System.err.println(\"Job execution failed: \" + e.getMessage());\n            e.printStackTrace();\n            System.exit(1);\n        }\n    }\n\n    /**\n     * Keyed process function with state optimized for ZGC:\n     * - Uses ValueState with short TTL to reduce heap retention\n     * - Avoids large object allocations in processElement\n     */\n    public static class ZGCStatefulProcessor extends KeyedProcessFunction {\n        private transient ValueState eventCountState;\n        private static final int MAX_EVENT_COUNT = 1000; // TTL replacement for simple state\n\n        @Override\n        public void open(Configuration parameters) {\n            ValueStateDescriptor stateDescriptor = new ValueStateDescriptor<>(\n                    \"event-count-state\",\n                    Integer.class\n            );\n            // Set TTL to 1 hour to automatically clean up state (reduces GC workload)\n            stateDescriptor.enableTimeToLive(\n                    org.apache.flink.api.common.state.StateTtlConfig\n                            .newBuilder(org.apache.flink.api.common.time.Time.hours(1))\n                            .cleanupFullSnapshot()\n                            .build()\n            );\n            eventCountState = getRuntimeContext().getState(stateDescriptor);\n        }\n\n        @Override\n        public void processElement(String event, Context ctx, Collector out) {\n            try {\n                Integer currentCount = eventCountState.value();\n                int newCount = (currentCount == null ? 0 : currentCount) + 1;\n                if (newCount > MAX_EVENT_COUNT) {\n                    eventCountState.clear(); // Explicit cleanup to avoid state bloat\n                    newCount = 1;\n                }\n                eventCountState.update(newCount);\n                out.collect(event + \",processed_count=\" + newCount);\n            } catch (Exception e) {\n                System.err.println(\"State processing failed for key: \" + ctx.getCurrentKey() + \", error: \" + e.getMessage());\n            }\n        }\n    }\n}\n

\n#!/bin/bash\n#\n# Flink 1.20 TaskManager JVM 21 ZGC Tuning Script\n# Validates environment, sets optimal ZGC parameters, and starts TaskManager\n# Exit codes: 0=success, 1=invalid env, 2=java version mismatch, 3=startup failure\n\nset -euo pipefail\n\n# Configuration variables (override via environment variables)\nFLINK_HOME=\"${FLINK_HOME:-/opt/flink}\"\nJAVA_HOME=\"${JAVA_HOME:-/usr/lib/jvm/java-21-openjdk-amd64}\"\nTASK_MANAGER_HEAP_SIZE=\"${TASK_MANAGER_HEAP_SIZE:-16g}\"\nZGC_CONC_GC_THREADS=\"${ZGC_CONC_GC_THREADS:-4}\" # Tune to 25% of vCPUs\nZGC_PARALLEL_GC_THREADS=\"${ZGC_PARALLEL_GC_THREADS:-2}\"\nMAX_DIRECT_MEMORY=\"${MAX_DIRECT_MEMORY:-8g}\" # RocksDB uses off-heap memory\n\n# 1. Validate environment prerequisites\nvalidate_env() {\n    echo \"Validating environment prerequisites...\"\n    if [[ ! -d \"$FLINK_HOME\" ]]; then\n        echo \"ERROR: Flink home directory $FLINK_HOME does not exist\"\n        exit 1\n    fi\n    if [[ ! -d \"$JAVA_HOME\" ]]; then\n        echo \"ERROR: Java home directory $JAVA_HOME does not exist\"\n        exit 1\n    fi\n    # Check Java version is 21+\n    JAVA_VERSION=$(\"$JAVA_HOME/bin/java\" -version 2>&1 | head -n 1 | cut -d'\"' -f2 | cut -d'.' -f1)\n    if [[ \"$JAVA_VERSION\" != \"21\" && \"$JAVA_VERSION\" != \"22\" ]]; then\n        echo \"ERROR: Java version must be 21 or higher, found $JAVA_VERSION\"\n        exit 2\n    fi\n    # Check available memory\n    TOTAL_MEM_KB=$(grep MemTotal /proc/meminfo | awk '{print $2}')\n    TOTAL_MEM_GB=$((TOTAL_MEM_KB / 1024 / 1024))\n    if [[ $TOTAL_MEM_GB -lt 32 ]]; then\n        echo \"WARNING: TaskManager recommended minimum 32GB RAM, found ${TOTAL_MEM_GB}GB\"\n    fi\n    echo \"Environment validation passed.\"\n}\n\n# 2. Set ZGC-optimized JVM arguments\nset_jvm_args() {\n    echo \"Setting ZGC JVM arguments for Flink TaskManager...\"\n    # Base ZGC args: enable ZGC, uncommit unused memory, concurrent compaction\n    ZGC_ARGS=(\n        \"-XX:+UseZGC\"\n        \"-XX:ConcGCThreads=$ZGC_CONC_GC_THREADS\"\n        \"-XX:ParallelGCThreads=$ZGC_PARALLEL_GC_THREADS\"\n        \"-XX:+ZGenerational\" # Enable generational ZGC (JVM 21+ feature)\n        \"-XX:ZAllocationSpikeTolerance=5\" # Handle allocation spikes from Flink state\n        \"-XX:+ZProactive\" # Proactively reclaim unused memory\n        \"-XX:+ZUncommit\" # Uncommit unused heap regions to reduce memory footprint\n        \"-XX:ZUncommitDelay=300\" # Wait 5 minutes before uncommitting\n        \"-Xms$TASK_MANAGER_HEAP_SIZE\"\n        \"-Xmx$TASK_MANAGER_HEAP_SIZE\"\n        \"-XX:MaxDirectMemorySize=$MAX_DIRECT_MEMORY\"\n        # Flink-specific JVM args to reduce GC pressure\n        \"-Djava.awt.headless=true\"\n        \"-Djdk.nio.maxCachedBufferSize=1024\" # Reduce direct buffer cache\n        \"-XX:ErrorFile=/var/log/flink/taskmanager-jvm-error.log\"\n        \"-XX:HeapDumpPath=/var/log/flink/heapdumps\"\n        \"-XX:+HeapDumpOnOutOfMemoryError\"\n    )\n    # Export as FLINK_TASKMANAGER_JAVA_OPTS for Flink to pick up\n    export FLINK_TASKMANAGER_JAVA_OPTS=\"${ZGC_ARGS[*]}\"\n    echo \"Set FLINK_TASKMANAGER_JAVA_OPTS: $FLINK_TASKMANAGER_JAVA_OPTS\"\n}\n\n# 3. Start TaskManager with error handling\nstart_taskmanager() {\n    echo \"Starting Flink TaskManager...\"\n    START_LOG=\"/var/log/flink/taskmanager-startup.log\"\n    if ! \"$FLINK_HOME/bin/taskmanager.sh\" start >> \"$START_LOG\" 2>&1; then\n        echo \"ERROR: Failed to start TaskManager. Check $START_LOG for details.\"\n        exit 3\n    fi\n    # Verify process started\n    sleep 5\n    if ! pgrep -f \"org.apache.flink.runtime.taskexecutor.TaskExecutor\" > /dev/null; then\n        echo \"ERROR: TaskManager process not found after startup.\"\n        exit 3\n    fi\n    echo \"TaskManager started successfully with ZGC tuning.\"\n}\n\n# 4. Main execution flow\nmain() {\n    validate_env\n    set_jvm_args\n    start_taskmanager\n    # Monitor GC logs for 10 seconds to verify ZGC is active\n    echo \"Verifying ZGC activation...\"\n    if tail -n 100 /var/log/flink/taskmanager.log | grep -q \"Using ZGC\"; then\n        echo \"SUCCESS: ZGC is active in TaskManager JVM.\"\n    else\n        echo \"WARNING: ZGC not detected in logs. Check JVM arguments.\"\n    fi\n}\n\n# Run main with error trapping\ntrap 'echo \"Script failed at line $LINENO\"; exit 1' ERR\nmain\n

\n#!/usr/bin/env python3\n\"\"\"\nZGC Metrics Exporter for Flink 1.20 TaskManagers\nCollects ZGC pause time, heap usage, and Flink event throughput metrics\nExposes metrics on port 9090 for Prometheus scraping\nRequires: prometheus-client, psutil, requests\n\"\"\"\n\nimport os\nimport time\nimport threading\nimport logging\nfrom prometheus_client import start_http_server, Gauge, Counter\nfrom prometheus_client.core import GaugeMetricFamily, CounterMetricFamily, REGISTRY\nimport psutil\nimport requests\nfrom requests.exceptions import RequestException\n\n# Configure logging\nlogging.basicConfig(\n    level=logging.INFO,\n    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'\n)\nlogger = logging.getLogger('zgc-metrics-exporter')\n\n# Metric definitions\nZGC_PAUSE_TIME_MS = Gauge(\n    'flink_zgc_pause_time_ms',\n    'ZGC pause time in milliseconds',\n    ['taskmanager_id']\n)\nZGC_HEAP_USED_BYTES = Gauge(\n    'flink_zgc_heap_used_bytes',\n    'ZGC heap used in bytes',\n    ['taskmanager_id']\n)\nZGC_HEAP_COMMITTED_BYTES = Gauge(\n    'flink_zgc_heap_committed_bytes',\n    'ZGC heap committed in bytes',\n    ['taskmanager_id']\n)\nFLINK_EVENT_THROUGHPUT = Gauge(\n    'flink_event_throughput_events_sec',\n    'Flink event throughput in events per second',\n    ['taskmanager_id']\n)\nFLINK_EVENT_DROP_COUNT = Counter(\n    'flink_event_drop_count_total',\n    'Total number of dropped events',\n    ['taskmanager_id']\n)\n\n# Configuration from environment variables\nFLINK_JMX_PORT = int(os.getenv('FLINK_JMX_PORT', '9100'))\nTASKMANAGER_ID = os.getenv('TASKMANAGER_ID', 'taskmanager-01')\nPOLL_INTERVAL_SECONDS = int(os.getenv('POLL_INTERVAL_SECONDS', '5'))\nPROMETHEUS_PORT = int(os.getenv('PROMETHEUS_PORT', '9090'))\nFLINK_REST_PORT = int(os.getenv('FLINK_REST_PORT', '8081'))\n\nclass ZGCMetricsCollector:\n    \"\"\"Collects ZGC metrics from Flink's JMX endpoint\"\"\"\n    def __init__(self, jmx_port: int, taskmanager_id: str):\n        self.jmx_port = jmx_port\n        self.taskmanager_id = taskmanager_id\n        self.jmx_url = f'http://localhost:{jmx_port}/jmx'\n\n    def collect_zgc_metrics(self) -> dict:\n        \"\"\"Fetch ZGC metrics from JMX. Returns dict of metric name to value.\"\"\"\n        metrics = {}\n        try:\n            # Fetch all ZGC-related MBeans\n            response = requests.get(\n                self.jmx_url,\n                params={'query': 'java.lang:type=GarbageCollector,name=ZGC'},\n                timeout=2\n            )\n            response.raise_for_status()\n            jmx_data = response.json()\n            if not jmx_data.get('beans'):\n                logger.warning(\"No ZGC MBeans found in JMX response\")\n                return metrics\n            # Parse ZGC metrics\n            for bean in jmx_data['beans']:\n                if 'LastGcInfo' in bean:\n                    # ZGC pause time (ZGC has no STW pauses, but this tracks concurrent phase time)\n                    last_gc = bean['LastGcInfo']\n                    if 'duration' in last_gc:\n                        metrics['pause_time_ms'] = last_gc['duration']\n                if 'CollectionCount' in bean:\n                    metrics['collection_count'] = bean['CollectionCount']\n                if 'CollectionTime' in bean:\n                    metrics['collection_time_ms'] = bean['CollectionTime']\n        except RequestException as e:\n            logger.error(f\"Failed to fetch JMX metrics: {e}\")\n        return metrics\n\n    def collect_heap_metrics(self) -> dict:\n        \"\"\"Fetch heap usage metrics from JMX\"\"\"\n        metrics = {}\n        try:\n            response = requests.get(\n                self.jmx_url,\n                params={'query': 'java.lang:type=Memory'},\n                timeout=2\n            )\n            response.raise_for_status()\n            jmx_data = response.json()\n            for bean in jmx_data.get('beans', []):\n                if 'HeapMemoryUsage' in bean:\n                    heap = bean['HeapMemoryUsage']\n                    metrics['heap_used_bytes'] = heap.get('used', 0)\n                    metrics['heap_committed_bytes'] = heap.get('committed', 0)\n        except RequestException as e:\n            logger.error(f\"Failed to fetch heap metrics: {e}\")\n        return metrics\n\nclass FlinkMetricsCollector:\n    \"\"\"Collects Flink event throughput metrics from REST API\"\"\"\n    def __init__(self, rest_port: int, taskmanager_id: str):\n        self.rest_port = rest_port\n        self.taskmanager_id = taskmanager_id\n        self.rest_url = f'http://localhost:{rest_port}/api/v1'\n\n    def collect_throughput_metrics(self) -> dict:\n        \"\"\"Fetch Flink task metrics for event throughput\"\"\"\n        metrics = {}\n        try:\n            # Get task managers list\n            response = requests.get(f'{self.rest_url}/taskmanagers', timeout=2)\n            response.raise_for_status()\n            taskmanagers = response.json().get('taskmanagers', [])\n            for tm in taskmanagers:\n                if tm.get('id') == self.taskmanager_id:\n                    # Get task metrics (throughput is numRecordsOutPerSecond)\n                    metrics['throughput'] = tm.get('numRecordsOutPerSecond', 0)\n                    metrics['dropped_events'] = tm.get('numRecordsDropped', 0)\n        except RequestException as e:\n            logger.error(f\"Failed to fetch Flink REST metrics: {e}\")\n        return metrics\n\ndef poll_metrics():\n    \"\"\"Poll metrics every POLL_INTERVAL_SECONDS and update Prometheus gauges\"\"\"\n    zgc_collector = ZGCMetricsCollector(FLINK_JMX_PORT, TASKMANAGER_ID)\n    flink_collector = FlinkMetricsCollector(FLINK_REST_PORT, TASKMANAGER_ID)\n    while True:\n        try:\n            # Collect ZGC metrics\n            zgc_metrics = zgc_collector.collect_zgc_metrics()\n            heap_metrics = zgc_collector.collect_heap_metrics()\n            # Update Prometheus metrics\n            if 'pause_time_ms' in zgc_metrics:\n                ZGC_PAUSE_TIME_MS.labels(taskmanager_id=TASKMANAGER_ID).set(zgc_metrics['pause_time_ms'])\n            if 'heap_used_bytes' in heap_metrics:\n                ZGC_HEAP_USED_BYTES.labels(taskmanager_id=TASKMANAGER_ID).set(heap_metrics['heap_used_bytes'])\n            if 'heap_committed_bytes' in heap_metrics:\n                ZGC_HEAP_COMMITTED_BYTES.labels(taskmanager_id=TASKMANAGER_ID).set(heap_metrics['heap_committed_bytes'])\n            # Collect Flink metrics\n            flink_metrics = flink_collector.collect_throughput_metrics()\n            if 'throughput' in flink_metrics:\n                FLINK_EVENT_THROUGHPUT.labels(taskmanager_id=TASKMANAGER_ID).set(flink_metrics['throughput'])\n            if 'dropped_events' in flink_metrics:\n                FLINK_EVENT_DROP_COUNT.labels(taskmanager_id=TASKMANAGER_ID).inc(flink_metrics['dropped_events'])\n            logger.info(f\"Updated metrics: throughput={flink_metrics.get('throughput', 0)} events/sec, heap_used={heap_metrics.get('heap_used_bytes', 0)/1e9:.2f}GB\")\n        except Exception as e:\n            logger.error(f\"Metrics poll failed: {e}\")\n        time.sleep(POLL_INTERVAL_SECONDS)\n\nif __name__ == '__main__':\n    logger.info(f\"Starting ZGC metrics exporter on port {PROMETHEUS_PORT}\")\n    # Start Prometheus HTTP server\n    start_http_server(PROMETHEUS_PORT)\n    # Start metrics polling in background thread\n    poll_thread = threading.Thread(target=poll_metrics, daemon=True)\n    poll_thread.start()\n    # Keep main thread alive\n    while True:\n        time.sleep(60)\n

# Test ConcGCThreads impact with JMH benchmark\njava -XX:+UseZGC -XX:ConcGCThreads=4 -jar flink-zgc-benchmark.jar\njava -XX:+UseZGC -XX:ConcGCThreads=2 -jar flink-zgc-benchmark.jar

state.backend: rocksdb\nstate.backend.rocksdb.incremental: true\nstate.backend.rocksdb.writebuffer.count: 6\nstate.backend.rocksdb.writebuffer.size: 64mb

flink_zgc_heap_committed_bytes{job=\"flink-taskmanager\"} - flink_zgc_heap_used_bytes{job=\"flink-taskmanager\"} > 8e9