Apache
Flink

Flink's runtime consists of a JobManager and one or more TaskManagers. The JobManager coordinates distributed execution while TaskManagers execute the actual operator code in slots.

Choose your state backend based on state size and latency requirements. RocksDB is recommended for production workloads with large state.

// Configure RocksDB state backend
StreamExecutionEnvironment env =
    StreamExecutionEnvironment.getExecutionEnvironment();

env.setStateBackend(new EmbeddedRocksDBStateBackend(true));

// Configure incremental checkpoints (ALWAYS enable)
RocksDBStateBackend rocksDB = new EmbeddedRocksDBStateBackend();
rocksDB.setNumberOfTransferThreads(4);

// Set checkpoint storage to distributed FS
env.getCheckpointConfig()
   .setCheckpointStorage("s3://flink-checkpoints/my-job");

// HashMapStateBackend — in-memory, fast, small state only
env.setStateBackend(new HashMapStateBackend());

// Increase heap accordingly in flink-conf.yaml
// taskmanager.memory.managed.fraction: 0.0
// taskmanager.memory.task.heap.size: 4gb

ValueStateDescriptor<Long> desc =
    new ValueStateDescriptor<>("counter", Long.class);
ValueState<Long> count =
    getRuntimeContext().getState(desc);

Checkpoints are Flink's mechanism for exactly-once fault tolerance. Configure them defensively — a failed checkpoint is a potential data loss event.

# Checkpoint interval — balance recovery time vs overhead
execution.checkpointing.interval: 60s

# Minimum pause between checkpoints
execution.checkpointing.min-pause: 30s

# Timeout — increase for large RocksDB state
execution.checkpointing.timeout: 10min

# Enable unaligned checkpoints for backpressured pipelines
execution.checkpointing.unaligned: true

# Keep 3 retained checkpoints
state.checkpoints.num-retained: 3

# Externalize on cancel
execution.checkpointing.externalized-checkpoint-retention:
  RETAIN_ON_CANCELLATION

Performance tuning in Flink involves JVM heap, managed memory, network buffers, and operator-level optimizations. Profile before tuning.

Use the Flink Connector ecosystem for certified, maintained integrations. Avoid custom source/sink implementations unless absolutely required.

KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers("broker:9092")
    .setTopics("orders")
    .setGroupId("flink-consumer")
    .setStartingOffsets(OffsetsInitializer.committedOffsets(
        OffsetResetStrategy.EARLIEST))
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();

env.fromSource(source,
    WatermarkStrategy
        .<String>forBoundedOutOfOrderness(Duration.ofSeconds(5))
        .withTimestampAssigner((e, t) -> parseTs(e)),
    "Kafka Orders Source");

Design for failure. Configure restart strategies that match your SLA and understand the trade-offs between recovery speed and checkpoint frequency.

Export Flink metrics to Prometheus. Alert on the following critical signals:

• ✓
  lastCheckpointDuration — Alert if >50% of checkpoint interval. Indicates state growth or I/O bottleneck.
• ✓
  numberOfFailedCheckpoints — Any non-zero value in production is a critical event.
• ✓
  currentInputWatermark — Monitor lag vs. wall-clock to detect watermark stalls from idle partitions.
• ✓
  backPressuredTimeMsPerSecond — Alert at >500ms/s. Indicates downstream saturation; scale or tune the slow operator.
• !
  numRecordsOutPerSecond — Establish baseline and alert on >40% drop within a 5-minute window.
• !
  JVM GC pause time — P99 GC pause >200ms causes checkpoint timeouts. Switch to ZGC on JDK 17+.

metrics.reporter.prom.factory.class:
  org.apache.flink.metrics.prometheus.PrometheusReporterFactory
metrics.reporter.prom.port: 9249
metrics.reporter.prom.interval: 15 SECONDS

# Include job-level labels for Grafana dashboards
metrics.scope.jm.job: <host>.jobmanager.<job_name>
metrics.scope.tm.task: <host>.taskmanager.<tm_id>.<job_name>.<task_name>

Flink clusters process sensitive data and must be hardened in production environments. At minimum, enforce the following controls.

Before promoting any Flink job to production, verify all items:

• ✓State backend configured to RocksDB with incremental checkpointing
• ✓Checkpoint storage points to durable object store (S3/GCS/ADLS)
• ✓Restart strategy set (not default fixed-delay)
• ✓Prometheus metrics reporter configured and scraped
• ✓Watermark strategy set with appropriate out-of-orderness bound
• ✓All state descriptors use registered serializers (no Kryo fallback)
• ✓State TTL configured on all keyed state
• ✓SSL enabled for internal cluster communication
• ✓Web UI / REST API not exposed to public internet
• ✓Load tested at 2× expected peak throughput in staging