Flink Kafka Connector Configuration for High-Throughput Pipelines

When a stream processing system must handle hundreds of thousands of events per second, the connector between Apache Flink and Apache Kafka is usually the part that decides whether you hit your throughput target. A poorly configured source or sink will bottleneck the whole Stream Processing pipeline, leaving compute idle while consumer lag climbs on the Kafka side. This guide covers production configuration patterns for the modern Flink Kafka connector that platform engineers and SREs can apply directly.

A note on API versions before anything else: the legacy FlinkKafkaConsumer and FlinkKafkaProducer classes were deprecated in favor of KafkaSource/KafkaSink (the source builder landed in Flink 1.14, the sink in 1.15) and were removed entirely in Flink 2.0. If you are starting a new pipeline, or running on any recent Flink, use the builder-based API shown throughout this article. Every code sample below uses org.apache.flink.connector.kafka.*. For a broader view of how this integration fits a larger analytics stack, see Integrating Apache Flink with Kafka for Advanced Analytics.

Understanding the Kafka source parallelism model

KafkaSource assigns Kafka partitions to Flink source subtasks. The effective parallelism of the source is bounded by the number of partitions in the subscribed topics: if you give the source operator more parallelism than there are partitions, the surplus subtasks sit idle with no partitions assigned. Unlike a plain consumer group, KafkaSource does not rely on Kafka’s group rebalance protocol to assign partitions — Flink discovers partitions and distributes them across subtasks itself, and it checkpoints the assigned offsets as part of Flink state. The group.id you set is used for offset reporting and ACLs, not for cooperative assignment.

Because the partition count caps source parallelism, partition planning is an operational decision you make before writing any Flink config. A common starting point is to set the source operator parallelism equal to (or an integer divisor of) the partition count.

# Create a throughput-oriented topic. Partition count caps source parallelism.
kafka-topics.sh --create \
  --bootstrap-server kafka-broker-1:9092,kafka-broker-2:9092,kafka-broker-3:9092 \
  --topic high-throughput-events \
  --partitions 48 \
  --replication-factor 3 \
  --config compression.type=lz4 \
  --config retention.bytes=107374182400 \
  --config segment.bytes=536870912

With 48 partitions, a source operator at parallelism 48 gives a clean one-partition-per-subtask mapping and no idle readers. When you scale up, increase partitions and parallelism together. Adding partitions to an existing topic is an online operation, but it breaks key-to-partition affinity for the default hash partitioner, so it changes ordering for any key that moves — plan it deliberately if you rely on per-key ordering.

There is a second, easy-to-miss consequence of adding partitions to a running job. Since the flink-connector-kafka 4.0.0 release (FLIP-288), dynamic partition discovery is enabled by default: the enumerator polls Kafka for new partitions every partition.discovery.interval.ms (default 5 minutes, aligned with metadata.max.age.ms) and assigns them to subtasks live, with no restart. Two operational points follow. First, partitions you add at runtime are picked up automatically — but with up to a five-minute lag, so a partition expansion is not instantaneous from Flink’s side. Second, newly discovered partitions are consumed from the earliest offset regardless of your configured starting strategy, which is deliberate: it guarantees the records written to a fresh partition before Flink noticed it are not skipped. If you need the old behavior — no live discovery — set the interval to 0 or a negative value with .setProperty("partition.discovery.interval.ms", "0").

Configuring the Kafka source for throughput

KafkaSource is built with a fluent builder. Arbitrary Kafka consumer properties — including the fetch-sizing knobs that matter most for throughput — are passed through with setProperty (or setProperties for a Properties object); valid keys are the standard ConsumerConfig keys plus Flink’s KafkaSourceOptions.

import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

KafkaSource<MyEvent> source = KafkaSource.<MyEvent>builder()
    .setBootstrapServers("kafka-broker-1:9092,kafka-broker-2:9092")
    .setTopics("high-throughput-events")
    .setGroupId("flink-high-throughput-consumer")
    .setStartingOffsets(OffsetsInitializer.committedOffsets(
        org.apache.kafka.clients.consumer.OffsetResetStrategy.EARLIEST))
    .setValueOnlyDeserializer(new MyEventDeserializationSchema())
    // Fetch sizing: defaults are conservative for high-volume topics.
    .setProperty(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, "52428800")           // 50 MB per fetch
    .setProperty(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, "10485760") // 10 MB per partition
    .setProperty(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, "1048576")            // 1 MB
    .setProperty(ConsumerConfig.RECEIVE_BUFFER_CONFIG, "1048576")             // 1 MB socket buffer
    .build();

DataStream<MyEvent> stream = env.fromSource(
    source, WatermarkStrategy.noWatermarks(), "kafka-source");

The starting-offsets choice above is deliberate, and it is a common foot-gun. The bare OffsetsInitializer.committedOffsets() form throws at runtime if the consumer group has no committed offset yet — which is exactly the state of a brand-new job, or one whose offsets aged out of __consumer_offsets. Passing OffsetResetStrategy.EARLIEST makes the source fall back to the earliest available offset when no commit exists (use LATEST if you would rather skip the backlog and only process new data), so the job starts cleanly on first deploy instead of failing.

fetch.min.bytes tells the broker to hold a fetch response until at least that much data is ready, which cuts the number of round trips when you are moving millions of small records (at the cost of a little latency, bounded by fetch.max.wait.ms, default 500 ms). fetch.max.bytes and max.partition.fetch.bytes raise the per-fetch ceiling so a single request can pull a large batch.

One caveat on max.poll.records: it is a valid ConsumerConfig key and KafkaSource does pass it through to the underlying consumer, but KafkaSource drives the consumer through its own split-reader loop rather than your application calling poll() directly. It still bounds how many records come back per internal poll, but it is not the primary throughput lever here — the fetch-byte settings above are. Tune it only if you see it in profiling, not by default.

Configuring the Kafka sink

KafkaSink replaces FlinkKafkaProducer. The record serializer is itself built with KafkaRecordSerializationSchema.builder(), and the fault-tolerance level is chosen with setDeliveryGuarantee.

import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.kafka.clients.producer.ProducerConfig;

KafkaSink<MyOutputEvent> sink = KafkaSink.<MyOutputEvent>builder()
    .setBootstrapServers("kafka-broker-1:9092,kafka-broker-2:9092")
    .setRecordSerializer(KafkaRecordSerializationSchema.<MyOutputEvent>builder()
        .setTopic("output-topic")
        .setValueSerializationSchema(new MyOutputSerializationSchema())
        .build())
    .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
    .setProperty(ProducerConfig.ACKS_CONFIG, "all")
    .setProperty(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4")
    .setProperty(ProducerConfig.BATCH_SIZE_CONFIG, "32768")        // 32 KB batches
    .setProperty(ProducerConfig.LINGER_MS_CONFIG, "5")             // wait up to 5 ms to batch
    .setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5")
    .setProperty(ProducerConfig.BUFFER_MEMORY_CONFIG, "67108864")  // 64 MB record accumulator
    .setProperty(ProducerConfig.MAX_BLOCK_MS_CONFIG, "60000")      // 60 s max block on send()
    .build();

stream.sinkTo(sink);

linger.ms adds a small, deliberate delay so more records accumulate per batch; larger batches compress better and cut request overhead. buffer.memory is the size of the producer’s record accumulator — when it fills (downstream Kafka can’t keep up), send() blocks up to max.block.ms, which is how the sink applies natural backpressure to the Flink job rather than dropping data.

On durability: acks=all together with min.insync.replicas on the broker is the only setting that survives a leader failure without data loss, so it is the right default for most pipelines despite a modest throughput cost. Use acks=1 only when you have explicitly decided you can tolerate the loss of unreplicated writes. The full set of knobs is in the Kafka producer configuration documentation.

For exactly-once output, switch to DeliveryGuarantee.EXACTLY_ONCE and add setTransactionalIdPrefix("...") with a value unique per application. This makes the sink use Kafka transactions that commit on checkpoint, so it must be paired with enabled checkpointing and read-committed consumers downstream. Critically, raise the broker’s transaction.timeout.ms (and the producer’s, which Kafka caps at the broker value, default 15 minutes) above your worst-case checkpoint interval plus recovery time — if a transaction times out before its checkpoint commits, you lose data. The cost is real: exactly-once buffers output until checkpoint completion, so your end-to-end latency floor rises to roughly the checkpoint interval. Reach for it only when downstream consumers genuinely cannot tolerate duplicates; otherwise AT_LEAST_ONCE plus idempotent downstream writes is cheaper and simpler.

Checkpointing and offset management

KafkaSource commits offsets back to Kafka on checkpoint completion, not on a fixed timer. The offsets in Kafka are for monitoring and tooling only; recovery always uses the offsets stored in Flink state. The practical consequence is that your checkpoint interval is also your offset-commit interval, and it governs how much data is replayed after a failure.

# Flink configuration excerpt (config.yaml / flink-conf.yaml)
execution.checkpointing.interval: 60s
execution.checkpointing.min-pause: 30s
execution.checkpointing.timeout: 10min
execution.checkpointing.max-concurrent-checkpoints: 1
execution.checkpointing.unaligned: true
state.backend.type: rocksdb
execution.checkpointing.dir: hdfs://namenode:8020/flink/checkpoints

Unaligned checkpoints let barriers overtake in-flight buffered data instead of waiting for it to drain, so a single slow or backpressured partition no longer stretches out the checkpoint and risks a timeout. They are restricted to max-concurrent-checkpoints: 1, which is why the two settings appear together. The RocksDB state backend (state.backend.type: rocksdb) is the choice for large keyed state because it keeps state on local disk rather than on the JVM heap.

Pick the interval by failure-replay tolerance: too short (every few seconds) hammers the state backend and inflates checkpoint overhead; too long means more reprocessing after a restart. 60 seconds is a reasonable starting point for a high-throughput job, tuned against your acceptable recovery time. The failure mode to watch for is checkpoints that start taking longer than the interval and begin to overlap or expire: if execution.checkpointing.timeout is hit repeatedly, offsets stop advancing in Kafka, lag climbs, and a restart replays everything since the last successful checkpoint. Alert on checkpoint duration as a fraction of the interval, not just on outright failures.

Note: in Flink 2.x the config file is config.yaml with standard YAML nesting; the flat flink-conf.yaml keys above still work on Flink 1.x and as flat keys. Use whichever matches your version.

Serialization and deserialization overhead

Serialization format has an outsized effect on throughput. JSON is readable but costs CPU to parse and inflates message size. For throughput-critical paths, prefer Avro with Schema Registry or Protobuf. Avro’s binary encoding commonly cuts payload size by roughly 40-60% versus JSON, which means less network I/O and less serializer CPU — though the exact ratio depends entirely on your schema, so measure it.

KafkaSource integrates with Confluent Schema Registry through ConfluentRegistryAvroDeserializationSchema, in the flink-avro-confluent-registry module. The two-argument forGeneric(Schema readerSchema, String registryUrl) overload is the correct form — the registry URL is a method parameter, not a Kafka consumer property.

import org.apache.flink.formats.avro.registry.confluent.ConfluentRegistryAvroDeserializationSchema;
import org.apache.avro.generic.GenericRecord;

KafkaSource<GenericRecord> source = KafkaSource.<GenericRecord>builder()
    .setBootstrapServers("kafka-broker-1:9092")
    .setTopics("avro-topic")
    .setGroupId("flink-avro-consumer")
    .setStartingOffsets(OffsetsInitializer.committedOffsets(
        org.apache.kafka.clients.consumer.OffsetResetStrategy.EARLIEST))
    .setValueOnlyDeserializer(
        ConfluentRegistryAvroDeserializationSchema.forGeneric(
            MyAvroSchema.getClassSchema(),
            "http://schema-registry:8081"))
    .build();

Schema Registry also gives you compatibility-checked schema evolution, which matters for long-running pipelines where producers and consumers are deployed independently. If you need a non-Confluent path, the Apache Avro specification documents the binary encoding.

Monitoring and operational metrics

KafkaSource exposes two layers of metrics. The first is the raw Kafka client metrics, re-registered by Flink under the KafkaSourceReader.KafkaConsumer group — for example records-consumed-rate, bytes-consumed-rate, and records-lag-max. The second is Flink-native source metrics, most importantly pendingRecords (the connector’s view of how many records are still available upstream, aggregated from per-partition lag) and currentEmitEventTimeLag. Export them through a metric reporter.

# Prometheus reporter (config.yaml / flink-conf.yaml).
# Note: factory.class + the *Factory class name is the current form.
metrics.reporter.prom.factory.class: org.apache.flink.metrics.prometheus.PrometheusReporterFactory
metrics.reporter.prom.port: 9249
# Optional filter. Syntax: <scope>[:<name>[,<name>][:<type>[,<type>]]]
metrics.reporter.prom.filter.excludes: "*:numRecordsIn*,numRecordsOut*"

A few exactness points that trip people up: the modern reporter is configured with metrics.reporter.<name>.factory.class pointing at PrometheusReporterFactory (the older .class + PrometheusReporter form is deprecated). Metric filtering uses filter.includes/filter.excludes with the scope-name-type pattern above — there is no filter.excludes “list of MBeans” syntax; it filters Flink metric scopes and names.

Three signals, read together, tell you what is actually wrong:

• records-lag-max trending up, backpressure high → a downstream operator is the bottleneck, not the source. Adding source parallelism won’t help; profile the slow operator or scale it.
• records-lag-max trending up, backpressure low → the source itself can’t keep up. Source parallelism (and therefore partition count) is too low, or fetch sizing is throttling reads.
• records-lag-max flat but high, pendingRecords flat → you are keeping pace with arrivals but never catching up on the backlog; you need a temporary parallelism bump or a one-off reprocess.

pendingRecords is Flink’s own accounting and is independent of the Kafka client metric, so a disagreement between the two usually points at stale client metrics rather than a real change. To confirm lag from the broker side, kafka-consumer-groups.sh --describe --group <id> shows committed-offset lag per partition — remember those commits only advance on checkpoint, so brief sawtooth lag at each checkpoint interval is normal, not a problem. The Kafka monitoring documentation covers the broker metrics around consumer-group health.

Tuning for specific workload patterns

Not every high-throughput pipeline wants the same trade-offs. The right configuration depends on whether you are optimizing for latency or for raw throughput, and on how evenly your keys are distributed. The two profiles pull the same knobs in opposite directions:

In both cases fetch.max.wait.ms (default 500 ms) bounds how long a fetch waits regardless of fetch.min.bytes, so it is the ceiling on the latency a large fetch.min.bytes can add.

For skewed key distributions, a handful of partitions carry most of the traffic, so the subtasks reading them become the bottleneck while others idle. After the source, insert rebalance() (full round-robin) or rescale() (round-robin within local subtask groups, cheaper on network) to spread records evenly across the downstream operator before any expensive processing. This only helps the stages after the redistribution — the source read itself is still partition-bound, so badly skewed partitions may still need a repartition at the producer or a larger partition count. The tell for source-side skew is uneven records-consumed-rate across subtasks while overall lag grows on just one or two partitions; no amount of downstream rebalance() fixes that, because the hot partition is read by a single subtask.

These are starting points, not universal constants. Apply them, watch records-lag-max, pendingRecords, and checkpoint duration under real load, and iterate from measured behavior rather than theoretical models. A configuration that is correct on your cluster and your data beats any nominal events-per-second figure copied from a guide.