Apache Kafka and Apache Flink integration

How Apache Flink connects to Apache Kafka topics

A team that needs Flink to read from or write to Kafka topics on a cluster it runs should start by understanding that Flink uses KafkaSource and KafkaSink, and that the integration depends on three things working together: offsets, checkpoints and transactions. This page covers the starting positions a source can use, how checkpoints drive offset commits, how exactly-once sinks rely on Kafka transactions, the transactional ID setup those sinks need, and the identity, testing and failure points that follow from them. Apache Kafka is licensed under the Apache License 2.0 (SPDX: Apache-2.0), and the project does not offer a paid edition, so the cost of this integration is the engineering and operating effort on your side. The connector behavior described here should be read alongside the Flink and Kafka documentation for the exact versions you plan to run.

The source and sink follow different rules, and the difference shapes the whole design. A KafkaSource takes its starting position from settings such as the offsets initializer, and Flink's own checkpoints hold the authoritative record of progress, so Kafka-side committed offsets mainly serve monitoring and tools that read consumer group state. Confirm in the connector documentation which offsets are committed back to Kafka and when, because a team that alerts on consumer lag will see different numbers depending on checkpoint interval and commit behavior. A KafkaSink that uses exactly-once delivery writes inside Kafka transactions that commit only when a Flink checkpoint completes. That design requires every producer transaction ID to be unique and stable across restarts, and it requires the brokers to accept the transaction settings you configure, so decide the naming scheme and the broker-side limits before the first job is deployed.

The team should also decide who owns each piece before any job goes to production. Kafka ACLs must allow the Flink principal to read the source topics, write the sink topics and use the transactional IDs and consumer groups the job depends on. With the KRaft StandardAuthorizer, resources without ACLs are limited to super users unless allow.everyone.if.no.acl.found is set, so a missing grant often appears as an authorization failure only after a job restarts from a checkpoint. Test failure paths deliberately: stop a TaskManager during a checkpoint, restart a job from a savepoint with a different parallelism, and confirm that downstream consumers using read_committed see no duplicate or aborted records. Those tests will tell you more about the integration than a successful first run.

Apache Kafka prerequisites for Flink jobs

A Flink job that reads and writes Kafka depends on decisions made on both sides of the connection. The Kafka cluster must be on a supported release in KRaft mode, the Flink runtime must carry a connector that provides KafkaSource and KafkaSink, and the network, certificates and accounts must line up before any job is submitted. Confirm each card with your own environment and the vendor documentation.

Supported Kafka release line

The Kafka downloads page lists 4.3.1, 4.2.2 and 4.1.2 as the supported releases at the time of research, and earlier versions are archived. Pick a release from that list and check the upgrade notes before committing, because the Kafka project issues bugfix releases only for supported lines. Record the chosen version, since the Flink connector and client libraries must be matched to it. Confirm the current list on the downloads page when you start the build.

KRaft mode on the cluster

Apache Kafka 4.0 supports only KRaft mode, and ZooKeeper-based clusters must migrate to KRaft before upgrading to 4.x. Flink jobs connect through bootstrap servers and do not care which metadata mode the brokers use, but the cluster you connect to must already be KRaft-based if you plan to stay on 4.x. Ask the platform team to confirm the metadata mode and the controller layout before you point a job at it.

Java runtime on brokers and clients

Since Kafka 4.0, brokers, Kafka Connect and tools require Java 17, and clients and Kafka Streams applications require at least Java 11. Flink task managers run the Kafka client inside their own JVM, so the Java version they run must also satisfy the Kafka client library you bundle. Compare the Java version required by your chosen Flink release, which you must confirm in Flink's documentation, against the Kafka client minimum before you build the image.

Flink release and Kafka connector

The job needs a Flink release whose Kafka connector provides KafkaSource for reading and KafkaSink for writing, with the exactly-once sink mode available. Connector versions are tied to Flink versions, so check the connector compatibility matrix in Flink's documentation before choosing both. Record the connector artifact, the Kafka client version it pulls in, and any Flink release that drops support for it, since upgrades on either side can break the job.

Checkpoint store and state backend

Checkpoints hold the Kafka offsets that a source commits and the transaction state an exactly-once sink relies on, so the checkpoint store must be durable and reachable from every task manager and the job manager. Choose the storage location, its retention, and the state backend with the platform team before the first run. Ask who owns the bucket or volume, how old checkpoints are cleaned up, and how a restored job is tested.

Network paths to brokers

Every task manager must reach every advertised broker address, not only the bootstrap address, because consumers and producers connect to partition leaders directly. Check DNS, firewall rules and any load balancer in front of the brokers, and confirm the advertised listeners resolve from the Flink network. Test from inside the Flink cluster with the same listener and port the job will use, since a test from a workstation proves little.

Certificates and job accounts

Kafka authenticates clients with SSL or SASL, and the documented SASL mechanisms are GSSAPI, PLAIN, SCRAM-SHA-256, SCRAM-SHA-512 and OAUTHBEARER. Decide which mechanism the job uses, create a dedicated principal for each job, and grant it only the topic, group and transactional ID permissions it needs under the StandardAuthorizer. Store truststores, keystores and credentials outside job code, and set a rotation owner, because an expired certificate stops the job without a clear error in Flink.

Apache Flink offset tracking in checkpoints

A Flink job does not resume from the offsets Kafka stores for its consumer group. It resumes from the source offsets recorded in its last successful checkpoint. Each KafkaSource keeps the partition offsets it has read as part of Flink's checkpointed state, and on restart the job seeks to those offsets. The consumer group offsets that Kafka holds for the job are a secondary record. Flink commits them back to Kafka when a checkpoint completes, so that external monitoring and lag tools can see progress. That commit is informational for Flink's own recovery. Teams that build dashboards or alerts on consumer group lag should know that the committed value trails the live position by up to one checkpoint interval, and that a job which has never completed a checkpoint has no committed position to show.

This design means the checkpoint interval and retention decide how much work is replayed after a failure. Records read after the last successful checkpoint are processed again on restart, so a longer interval means a larger replay window and more duplicate output for any sink that is not transactional. A shorter interval reduces replay but adds checkpointing overhead on the job. Retention matters too: if the job is stopped for longer than the topic's retention period, the offsets in its checkpoint may point to data Kafka has already deleted, and the job will either fail or skip ahead depending on its offset reset behavior. Confirm the retention period of each source topic against the longest outage you plan to recover from, and confirm the checkpoint retention policy in your Flink deployment.

The decisions to record before the build are the checkpoint interval, the number of retained checkpoints, the expected maximum recovery time, and the topic retention for each input. Test them with a planned restart: stop the job, let a known number of records arrive, restart it from the last checkpoint, and compare the output against the expected results. Check the Flink and Kafka connector documentation for the exact recovery behavior of the version you run, since this is where the offset semantics are defined.

Configuring the Flink to Kafka integration in sequence

This sequence wires a Flink job to Kafka topics so that reads resume from known offsets, writes commit only when Flink checkpoints complete, and downstream consumers see only committed output. Each step ends with a check that confirms the piece works before the next one depends on it.

Create the topics with planned partition counts
The Kafka administrator creates the source, sink and any dead-letter topics before the job exists, with partition counts set from the throughput and parallelism plan. Partition count caps how many parallel source readers can be assigned work, and changing it later affects key ordering, so confirm the number with the team that owns the job. Set replication and retention to the values in your platform standard. Done when the topic descriptions show the expected partition count, replication factor and retention settings on every broker in the cluster.
Create the Kafka principal and ACLs
The administrator creates a dedicated principal for the Flink job and grants it read on the source topics and the consumer group, write on the sink topics, and access to the transactional ID prefix that the sink will use. Under the StandardAuthorizer, resources without ACLs are restricted to super users unless the cluster has allow.everyone.if.no.acl.found set to true, so a missing grant will surface as an authorization failure rather than silent access. Done when a test client authenticated as that principal can produce to the sink topic and consume from the source, and is denied on a topic outside its grant.
Configure the KafkaSource starting position
The Flink developer sets the KafkaSource to a starting position that matches the recovery intent: committed group offsets for normal restarts, or an explicit timestamp or earliest offset for a backfill. The choice matters because it determines what the job reprocesses on first start. Confirm the source's group ID matches the group granted in the previous step. Done when a dry run against a small test topic reads the expected first record and the job's log shows the chosen starting strategy.
Enable checkpointing and confirm the checkpoint store
The Flink developer enables checkpointing and points the checkpoint storage at a durable location, such as a distributed file system or object store that survives TaskManager loss. The KafkaSource commits offsets as part of checkpoints, so checkpoint interval and storage durability together determine how much work is replayed after a failure. Confirm the storage path is not local to a single node. Done when a completed checkpoint is listed in the checkpoint store and the job's checkpoint metrics show successful completions.
Configure the KafkaSink delivery guarantee
The Flink developer sets the KafkaSink delivery guarantee to exactly-once for transactional output, and sets a transactional ID prefix that is unique to this job and matches the ACL granted earlier. Transaction timeouts on the broker must be compatible with the checkpoint interval, or open transactions may be aborted during recovery. Ask the platform team to confirm the broker transaction settings before deployment. Done when the job configuration shows the guarantee and prefix, and the prefix appears in the transactional ID grant.
Deploy the job and observe checkpoint behavior
The Flink operator deploys the job to the target cluster and watches the first checkpoints complete. The sink should hold its output in open transactions until each checkpoint completes, then commit them. Check the job's exception log and the broker's transaction state for aborted or stuck transactions. Done when at least one checkpoint completes without errors and the job reports the sink as running with no repeated restarts.
Verify with a read_committed consumer
The tester produces a marker record through the job's input and waits for the next checkpoint to complete. A consumer configured with isolation.level set to read_committed should see the record only after that checkpoint, and should not see records from transactions that were aborted. Repeat the check after forcing a restart of the job so the test covers recovery. Done when the marker appears only after checkpoint completion, and a restart produces no duplicate committed records in the test output.

Choosing a starting offset and delivery guarantee by situation

A new pipeline has no saved state and no earlier output, so the decision is where the first read begins. Start from the earliest retained offset when every record still in the topic must be processed, or from the latest offset when only new events matter. Checkpointing is still needed for recovery, and the sink guarantee should follow from what downstream readers can tolerate.

Starting position: OffsetsInitializer.earliest() or OffsetsInitializer.latest() on the KafkaSource builder.
Delivery guarantee: AT_LEAST_ONCE is a common starting point when downstream consumers can handle duplicates. Use EXACTLY_ONCE when they cannot.
Isolation: consumers of an exactly-once sink need isolation.level=read_committed. The default read_uncommitted returns records from open or aborted transactions.
Main risk: starting from the earliest offset on a long-retained topic can produce a large first run that overloads the sink.

Before the first run, confirm the topic's retention window and the isolation setting of every consumer that reads the output.

When checkpointing is enabled, the KafkaSource stores its partition offsets in Flink state and restores them after a failure or a savepoint restore. The configured starting position matters only when a job starts without state. On the sink side, an exactly-once guarantee ties each Kafka transaction to a checkpoint, so recovery either commits or aborts the writes that belong to it.

Starting position: restored checkpoint or savepoint offsets take precedence. OffsetsInitializer.committedOffsets() serves as the fallback for a job that starts without state.
Delivery guarantee: EXACTLY_ONCE on the KafkaSink, with checkpointing enabled for the job.
Isolation: read_committed consumers see each record once its checkpoint commits, so they wait for the transaction to finish.
Transactional ID prefix: set setTransactionalIdPrefix to a value unique to this job. Two jobs with the same prefix can fence each other's transactions.
Main risk: Flink's documentation warns that data can be lost if the broker's transaction timeout is shorter than the checkpoint interval plus the restart time. Confirm the broker's transaction timeout settings, such as transaction.max.timeout.ms, against your checkpoint configuration.

Check that the transaction timeout settings and the transactional ID prefix match the job's checkpoint design before you rely on a restart.

A backfill reprocesses history from a chosen point while live processing continues. The starting position is usually a timestamp, which Kafka resolves to offsets per partition, or an explicit offset list when the team already knows the positions. Run the backfill as its own job with its own output path so historical writes do not interleave with live results without a plan.

Starting position: OffsetsInitializer.timestamp(...) starts each partition at its first record at or after the given time. OffsetsInitializer.offsets(...) pins exact positions.
Delivery guarantee: EXACTLY_ONCE when the backfill writes to a topic that live consumers read. AT_LEAST_ONCE is acceptable when the target store deduplicates by key.
Isolation: consumers of the backfill output need read_committed. Reconcile record counts against the live output before the results are published.
Transactional ID prefix: use a prefix distinct from the live job so that the backfill cannot fence live writes.
Main risk: the requested past point may no longer exist if retention has removed it, and timestamp lookups depend on what producers wrote into the record timestamp field.

Before you start, confirm the retention window, the producers' timestamp semantics and the output topic's consumers.

Downstream readers that must never see aborted writes depend on two settings working together: an exactly-once sink on the Flink side and read_committed on every consumer. An exactly-once sink alone does not protect readers that run with the default isolation level, and the guarantee holds only for consumers that filter out aborted records.

Starting position: not a sink setting. Choose the source position from the job's situation (new, resumed or backfill).
Delivery guarantee: EXACTLY_ONCE on the KafkaSink, with checkpointing enabled and a stable transactional ID prefix.
Isolation: every consumer group must set isolation.level=read_committed, including connectors and tools owned by other teams.
Access control: grant the Flink principal the ACLs its transactional IDs require, on the TransactionalId resource type and for idempotent writes. Confirm the exact operations in the Kafka authorization documentation for your version.
Main risk: an open transaction holds back the last stable offset for its partition, so read_committed consumers pause on that partition until the transaction completes. Slow checkpoints therefore show up as consumer lag.

Before go-live, ask each downstream team to confirm its isolation setting and to agree on acceptable lag during checkpoints.

Kafka settings and Flink configuration that must agree

This table maps each Kafka concept to the Flink setting that controls it and the failure that follows when the two sides disagree. Read it before the first deployment, then again whenever a job is copied, renamed or moved to a new environment. Confirm exact configuration key names against the Flink Kafka connector documentation for the Flink release you run.

Kafka concept	Flink setting or decision	What goes wrong when they disagree
Consumer group identity	Set the group ID on KafkaSource so each job has a stable, unique group that matches its offset history	Reusing a group ID across jobs makes them share offsets, so one job can skip or re-read records meant for the other
Transactional ID prefix	Give each KafkaSink a transactional ID prefix unique to the job, and keep it stable across restarts	Two jobs with the same prefix fence each other's producers, causing repeated aborts and stalled checkpoints
Transaction timeout	Set the sink's transaction timeout to cover the longest checkpoint interval plus recovery time, and keep it within the broker's transaction.max.timeout.ms	A timeout shorter than the checkpoint interval aborts open transactions, while one above the broker limit fails the producer at startup
Isolation level for downstream readers	Set downstream consumers to read_committed when the sink writes with exactly-once semantics	Readers on read_uncommitted see records from aborted or still-open transactions, so output looks duplicated or partial
Partition-to-subtask assignment	Size source parallelism against the topic partition count and let the connector assign partitions to subtasks	Parallelism above the partition count leaves idle subtasks, while changing partition count later shifts assignments and can disturb state expectations
Serializers and key handling	Pick the key and value serializers on KafkaSink to match the consumer's deserializers and the key used for partitioning	A changed key or format silently moves records to other partitions or breaks deserialization in downstream consumers
Topic naming per environment	Parameterize topic names per environment and pass them into both source and sink configuration from one place	A job pointed at a production topic from a test environment writes real output to the wrong system, and the mismatch is often found late

Securing the Flink job's access to Kafka

A Flink job that reads and writes Kafka is a client like any other, so it needs encryption, a named identity and permissions scoped to the topics and transactional IDs it uses. The cards below cover the security options Kafka documents, how to design the job's ACLs, and how to keep credentials out of job configuration.

Encrypt client traffic with TLS

Kafka encrypts traffic between clients, brokers and tools with SSL/TLS, and security is optional, so a cluster can run unsecured or mix secured and unsecured listeners. For a Flink job, configure the client to use a TLS listener and confirm the broker listener that the job's bootstrap servers resolve to is the secured one. Check that the truststore the job uses trusts the broker certificate chain, and plan certificate rotation before the job runs for months.

Choose a SASL mechanism for the job

Kafka documents SASL with GSSAPI (Kerberos), PLAIN, SCRAM-SHA-256, SCRAM-SHA-512 and OAUTHBEARER for authenticating clients. SCRAM stores credentials in the cluster and suits a dedicated job account, while OAUTHBEARER fits organizations that already issue tokens from an OIDC provider. Pick one mechanism per listener, and confirm with your security team which mechanisms your broker configuration and client libraries support before the job design is fixed.

Design ACLs for the job principal

In KRaft clusters, the StandardAuthorizer stores ACLs in the cluster metadata log. Resources without ACLs are restricted to super users unless allow.everyone.if.no.acl.found is set to true, so leaving it false is the safer default: the job gets access only where you grant it. Grant the job's principal read on its input topics, write on its output topics, and read on its consumer group. Confirm the exact operations your Flink version needs against the authorization documentation before you finalize the list.

Grant transactional ID access deliberately

An exactly-once Flink sink writes through Kafka transactions, which means the job's principal needs access to its transactional ID as well as to the topics. Because the transactional ID is a separate resource from the topic, a missing grant can look like a topic permission problem during debugging. Confirm the resource type and operations that govern transactional IDs in the authorization documentation, and write the grant with the same prefix or name your Flink sink uses.

Use one principal per job

Give each Flink job its own Kafka principal rather than sharing one account across jobs. A dedicated principal lets you revoke or rotate one job's credentials without stopping others, and it makes the audit trail show which job read or wrote a topic. It also keeps ACLs narrow, since each principal only needs the topics, consumer groups and transactional IDs of that one job.

Keep credentials out of job configuration

SASL passwords, keystore passwords and client secrets should not appear in Flink configuration files, job JARs or deployment manifests in plain text. Reference them from your secret store and inject them at deployment time, and confirm that Flink's own logs and the Flink web interface do not echo the client properties. Ask who can read the secret, how rotation reaches running jobs, and whether a restart is needed after a rotation.

Common Flink and Kafka failures and how to fix them

This table maps the failures teams most often hit when Flink reads from or writes to Kafka to their likely causes, fixes and checks. Read it by symptom: find what the downstream reader or job is showing, then work through the cause and check column before changing settings.

Symptom	Likely cause	Fix and check
Records invisible to downstream readers	Exactly-once sink transactions are still pending or were aborted, and the reader uses read_committed isolation	Confirm consumers set isolation.level to read_committed. Check that checkpoints complete, since transactions commit only on checkpoint completion, and inspect aborted transactions after failures
Transactional writes fail after a restart	The transactional ID is reused by a new job instance or a previous instance is still fenced or open	Give each sink a unique transactional ID prefix per job, and confirm the prefix is kept stable across restarts from the same checkpoint. Check broker logs for fencing errors
Large replay of windows after recovery	Checkpoints are infrequent, time out or fail, so the restored offsets are old	Review checkpoint interval, duration and failure count in the Flink UI. Tune checkpoint timeout and state backend size, and confirm offsets are committed on checkpoint completion
Authorization errors on the consumer group or transactional ID	ACLs do not grant Describe, Read or Write on the group or the transactional ID resource, or resources without ACLs are restricted to super users	Grant the Flink principal access to the consumer group, topic and TransactionalId resources the job uses. Test with a kafka-acls listing and a failed-job restart in a staging cluster
Subtasks sit idle with no records	The topic has fewer partitions than the source parallelism, so some source subtasks receive no partitions	Set source parallelism at or below the partition count, or add partitions if throughput needs more readers. Check per-subtask record counts in the Flink dashboard
TLS handshake failures on connect	The client trusts the wrong CA, uses a listener that does not match the protocol, or the endpoint is a plaintext listener	Match the client security protocol and truststore to the advertised listener for that bootstrap address. Check broker logs for SSL handshake errors and confirm the listener name and port

Questions teams ask before building Flink on Kafka

What drives the cost and effort of a Flink on Kafka integration?

The main drivers are the number of Flink jobs, the size of their state, whether you need exactly-once delivery, the security model on both systems, and how many downstream consumers depend on the output topics. Each job needs its own source and sink wiring, offset handling and failure testing, so job count multiplies effort. Large state increases checkpoint tuning and recovery testing. Exactly-once adds transaction configuration and more testing. Ask any partner for a quote that breaks effort down by job, by security integration and by test scope, and states which parts are one-time build work and which recur through upgrades and monitoring.

How long does a first Flink on Kafka integration take?

Duration depends on how many jobs are in scope, how mature the existing Kafka and Flink clusters are, how quickly access and certificates are issued, and how many environments must be validated before production. A single job with plaintext access to an existing cluster is far simpler than several secured jobs that write exactly-once into shared topics. Ask for a timeline tied to a written scope, with dependencies on access approvals listed separately. For the Flink side of the work, the Apache Flink documentation covers the connector behavior you will need to plan around.

What must our team provide before the integration starts?

Plan to provide Kafka admin access or an admin contact who can create topics, principals and ACLs, access to the Flink cluster or its deployment pipeline, and the certificates or SASL credentials each side uses. You also need a test topic plan that names the source and sink topics, partition counts, retention and the consumer groups that will read them. Confirm in writing who approves production changes and where secrets are stored. Missing any of these usually pushes work into later phases, so list them in the statement of work with owners and dates.

Can an internal team build this integration alone?

An internal team with working knowledge of both Kafka and Flink can often build a single job alone, especially if the work is limited to plaintext or basic TLS. Outside help is most useful for exactly-once sink design, transactional ID planning, offset starting points during migrations, SASL or mTLS wiring across both platforms, and failure testing that forces checkpoint recovery and broker restarts. Teams also tend to need help when several jobs share a Kafka cluster and quota or ACL rules must be agreed. Review the Flink connector documentation with your engineers before deciding which parts to keep in-house.

Does exactly-once delivery change the cost of running the integration?

Yes, exactly-once adds recurring operational work, even though it does not change the license cost, since Apache Kafka is licensed under Apache-2.0 with no paid edition from the project. Transactional sinks depend on checkpoint health, transaction timeouts and unique transactional IDs per sink instance, so operators must monitor open transactions and checkpoint duration. Downstream consumers must read with the isolation level that hides uncommitted records. Ask a partner to specify the monitoring, alerting and runbook work that exactly-once requires and to name which of those tasks your team will own after handover.

Which Kafka version and Java runtime should we plan for?

Plan on a currently supported Kafka release. The Kafka downloads page lists 4.3.1, 4.2.2 and 4.1.2 as supported releases, and Kafka 4.0 supports only KRaft mode. Since Kafka 4.0, brokers, Kafka Connect and tools require Java 17, while clients and Kafka Streams applications require at least Java 11. Confirm the Flink connector version you need is compatible with your chosen Kafka client libraries and Java runtime before the build starts, using the Apache Flink documentation as the authority for connector compatibility.

How do security settings affect the Flink and Kafka setup?

Kafka supports SSL and SASL for client and broker authentication, with GSSAPI, PLAIN, SCRAM-SHA-256, SCRAM-SHA-512 and OAUTHBEARER as the documented SASL mechanisms. In KRaft clusters, the StandardAuthorizer stores ACLs in the metadata log, and resources without ACLs are restricted to super users unless allow.everyone.if.no.acl.found is set to true. Each Flink job principal needs explicit read, write and transactional permissions on its topics and consumer groups. Confirm the exact client properties and credential storage method with your Flink deployment model in the Flink connector documentation before issuing credentials.

How DeployOpen scopes and delivers this integration

DeployOpen plans and implements Kafka and Flink wiring inside infrastructure you control, starting from a written scope that names the topics, offset rules, checkpoint storage and delivery guarantees. Engagements are fixed-scope, so the deliverables and the verification steps are agreed before any build work begins.

How we scope

Inventory the Kafka topics, consumer groups and sinks that the Flink jobs will read and write
Record the required starting offsets, replay expectations and retention windows for each job
Agree whether sinks need exactly-once delivery, and define the transactional ID scheme and its ownership
Confirm the Kafka version, authentication method and ACL model the integration must work with

How we build

Configure KafkaSource and KafkaSink in the Flink job code and deployment manifests for your environment
Set checkpoint intervals and storage, and verify that offsets commit as the checkpoints complete
Create transactional ID and ACL settings, then test sink behavior under failure and restart
Run the test plan in a non-production environment that mirrors your Kafka and Flink versions

How we hand over

Deliver the job code, configuration, deployment files and an operations runbook for your team
Document the offset, checkpoint and transaction settings, with the reason each value was chosen
Walk your engineers through restart, savepoint and failed-job recovery procedures before handover

How we support

Support is available as a retainer for upgrades, incident response and changes to jobs or topics
Revisit the Flink and Kafka version pairing when either project publishes a supported release
Confirm the retainer scope, response expectations and test cadence in the written agreement before work starts