Apache Flink and Kafka Integration

Apache Flink is a distributed processing engine for stateful computations over bounded and unbounded data streams. It can consume events as they arrive, maintain application state, use event time, recover from failures, and continuously publish results.

Apache Kafka is a distributed event-streaming platform. Producers append records to partitioned topics, Kafka retains those records, and consumers read them at their own pace. Kafka is commonly the durable transport layer in a streaming architecture, while Flink performs the computation:

applications -> Kafka topics -> Flink job -> Kafka topics / databases / APIs

The two systems solve different problems. Kafka stores and transports event logs. Flink transforms those logs into results by filtering, enriching, aggregating, joining, and detecting patterns. Their integration is especially useful when a computation must preserve state and remain correct after a process, machine, or network failure.

Glossary #

Stream Types #

Stream type will affect Job code primarily through source configuration. Downstream DataStream code can often remain identical.

// Unbounded: continuously discovers and consumes Kafka records
KafkaSource<Event> liveSource = KafkaSource.<Event>builder()
        .setTopics("events")
        .setStartingOffsets(OffsetsInitializer.committedOffsets())
        .build();

// Bounded: finishes at specified partition offsets
KafkaSource<Event> replaySource = KafkaSource.<Event>builder()
        .setTopics("events")
        .setStartingOffsets(OffsetsInitializer.offsets(startOffsets))
        .setBounded(OffsetsInitializer.offsets(endOffsets))
        .build();

Both can feed the same pipeline:

env.fromSource(source, watermarks, "events")
   .keyBy(Event::customerId)
   .process(new CustomerProcessor())
   .sinkTo(sink);

Boundedness also affects what code makes sense:

• Unbounded streams need windows or another emission policy for operations such as counts and aggregates. • A global “sort everything” cannot finish on an unbounded stream. • Bounded jobs can produce final results after end-of-input. • Continuous file monitoring makes a normally bounded file source unbounded.

You can additionally select runtime behavior:

env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

Source boundedness determines whether the job can terminate; runtime mode determines how Flink schedules and exchanges its data.

Why Use Flink with Kafka? #

• Flink works fine without kafka. Unbounded stream frequently means file/files on HFDS.
• Kafka works fine without Flink. We got good old custom deployment of Kafka consumers.

A basic Kafka consumer can read and transform records, but a real streaming application often needs more:

• keyed state that may be larger than one machine’s memory;
• event-time windows for late or out-of-order events;
• consistent recovery across input offsets, operator state, and output;
• rescaling without manually redistributing state;
• joins between streams or between streams and reference data;
• asynchronous checkpoints without stopping normal processing;
• SQL over continuously arriving records; and
• operational metrics, savepoints, and controlled upgrades.

Flink supplies these capabilities as part of one execution model. Kafka provides replayable input and durable output, allowing a failed Flink job to resume from a consistent point rather than starting over or guessing which records were processed.

Flink is not automatically better than a custom Kafka consumer. A plain consumer is often simpler when each record is independent and processing is only:

consume -> perform one stateless operation -> publish or store

The custom service must otherwise implement or integrate state persistence, offset coordination, backpressure, retries, event-time handling, rescaling, and recovery. Flink becomes valuable when several of those concerns are part of the computation. The trade-off is the operational cost of running or using a Flink cluster instead of deploying an ordinary service.

Best-Fit Mental Model and Use Cases #

Think of a Flink job as a distributed, continuously evaluated database query over event streams:

unbounded input -> transformations and durable state -> continuous output

Kafka retains and transports the event history. Flink continuously computes derived state and results from that history. This model is a good fit for:

• fraud detection: detect a burst of card transactions across distant locations within a rolling time interval;
• real-time analytics: continuously update sales, active-user, and conversion metrics;
• streaming ETL: validate orders, enrich them with customer or product data, and publish analytics-ready records;
• recommendations: update a shopper’s interests from searches, clicks, and purchases;
• monitoring: correlate service events and alert when a rolling failure rate crosses a threshold;
• deduplication: remember event identifiers so retried payments or notifications are not processed twice;
• sessionization: group web activity until a period of inactivity closes the session;
• stream joins: associate an advertisement click with a later purchase, including late or out-of-order events;
• logistics: combine GPS updates and route data to update estimated arrival times; and
• IoT processing: aggregate sensor readings and detect abnormal patterns.

Flink Without Kafka and Its Streaming-First Design #

Flink does not require Kafka. Kafka is one possible source and sink among systems such as Pulsar, Kinesis, database change-data-capture feeds, files, Iceberg tables, and application-specific connectors:

database CDC -> Flink -> Iceberg
Kinesis      -> Flink -> database
HDFS files   -> Flink -> HDFS files
IoT events   -> Flink -> alerts

Without Kafka, Flink still overlaps substantially with Apache Spark, particularly for bounded SQL and ETL workloads. Their main difference is their center of gravity rather than an absolute capability boundary:

Finite file transformations and data-lake analytics are often a natural Spark workload. Flink remains distinct without Kafka when the application needs continuous state, fine-grained time semantics, low latency, or long-lived event-driven behavior.

Streaming Capabilities and How Flink Implements Them #

Flink makes state, time, recovery, and flow control runtime concerns rather than requiring every source consumer to implement them independently:

Managed state is logically attached to an operator and, for keyed state, to the current application key. Runtime state may reside on the JVM heap or in an embedded state backend, while checkpoints copy a consistent snapshot to durable storage. On recovery or rescaling, Flink restores and redistributes that state rather than requiring the application to rebuild it manually.

Event-time timers fire when the watermark passes their timestamp, not merely when wall-clock time advances. A window can retain its result for an allowed lateness interval and update it when a delayed event arrives; records later than that policy can be routed to a side output. Processing-time timers instead follow the TaskManager’s clock and are useful when wall-clock behavior is the actual requirement.

Core Flink Concepts #

Before connecting the systems, it helps to understand Flink’s model.

Jobs and Operators #

A Flink application defines a graph of operators:

source -> transform -> partition -> window / process -> sink

Flink turns this graph into a distributed job. Each operator can have multiple parallel subtasks. Records flow continuously through the graph rather than waiting for an entire input data set to finish.

In DataStream code, an operator is a logical processing step and a transformation is the API operation that adds that step to the job graph. The terms are often used interchangeably. Flink may chain compatible operators into one runtime task for efficiency, so operators do not necessarily map one-to-one to threads, processes, or network exchanges.

Source and Sink Operators #

A source introduces records into the job. It also defines serialization, partition discovery, boundedness, and the positions from which recovery continues.

Production connectors should normally use fromSource. Collection and in-memory sources are mainly useful for examples and tests.

A sink sends results out of Flink.

Sources and sinks form the job’s external boundaries. A sink’s delivery guarantee depends on both the connector and checkpoint configuration. Writing to an external system inside map, process, or another transformation bypasses the sink protocol and is not automatically coordinated with checkpoints.

Basic Transformation Operators #

Basic transformations handle records independently and usually require no managed state:

Use map for parsing or enrichment already available in memory, flatMap for tokenization or validation that may emit multiple results, and filter for discarding unwanted records. returns describes a transformation’s type; it does not process records itself.

Partitioning Operators #

Partitioning determines which downstream subtask receives each record. It may introduce a network exchange:

keyBy is the most important partitioning operation because it creates a KeyedStream, enabling keyed state, timers, keyed windows, and keyed aggregations. It is not the same as Kafka partitioning: Kafka assigns records to source partitions before Flink reads them, while keyBy repartitions records inside the Flink job.

Aggregation Operators #

Aggregations combine multiple records into a continuously updated result. Common APIs are:

Unwindowed aggregations on a KeyedStream update indefinitely and therefore retain one aggregate per key. Windowed reduce and aggregate limit the aggregation to each window. Prefer aggregate when the accumulator should be different from both the input and result, such as maintaining a sum and count to produce an average.

Window Operators #

Windows group an unbounded stream into finite buckets for computation. Frequently used window assigners are:

After keyBy, use window(...); without a key, use windowAll(...). Frequently used window computations are reduce, aggregate, and process. Incremental reduce or aggregate is generally more efficient than process, which may need access to all records in a window. They can be combined so that Flink incrementally aggregates records and a ProcessWindowFunction adds window metadata to the final result.

Process Operators #

Process functions are Flink’s low-level escape hatch for logic that needs managed state, timers, timestamps, watermarks, or side outputs:

The commonly used API call is process(...). Side outputs are declared with an OutputTag and emitted through the function context, which is useful for late events, invalid records, or separately routed alerts. Use a simpler transformation when state and timers are unnecessary.

Multi-Stream and Join Operators #

These operators combine related streams:

union interleaves records but does not join them. connect allows shared logic or state across two inputs but also does not correlate records automatically. Use intervalJoin, a windowed join, or a KeyedCoProcessFunction when records must be matched by key and time.

Asynchronous I/O Operators #

Asynchronous I/O lets many external requests be in flight without blocking a task thread:

Both use an AsyncFunction or RichAsyncFunction plus a timeout and capacity. They are commonly used for remote lookups that cannot be represented as a stream or broadcast state. The external client must actually be asynchronous; wrapping a blocking call does not make it nonblocking. Define timeout, retry, ordering, and replay behavior explicitly.

DataStream and Its Specialized Types #

org.apache.flink.streaming.api.datastream.DataStream<T> is the base Java abstraction for a stream whose records have type T. It is a handle to a logical transformation in the job graph, not an in-memory collection of all records. Calling methods such as map, keyBy, or sinkTo adds transformations to that graph and returns another API object.

Flink returns specialized DataStream subtypes when the next valid operations depend on how the stream was produced or partitioned:

The hierarchy can be summarized as:

DataStream<T>
├── SingleOutputStreamOperator<T>
│   └── DataStreamSource<T>
├── KeyedStream<T, K>
├── SideOutputDataStream<T>
└── CachedDataStream<T>

The declared Java type determines which fluent methods are visible. For example:

SingleOutputStreamOperator<Purchase> purchases = raw
        .map(OrderPipeline::parsePurchase)
        .name("parse-purchase");

KeyedStream<Purchase, String> byCustomer =
        purchases.keyBy(Purchase::customerId);

SingleOutputStreamOperator<Purchase> totals = byCustomer
        .reduce(OrderPipeline::addAmounts)
        .uid("customer-totals");

name and uid are visible on SingleOutputStreamOperator, while keyed reduce is visible on KeyedStream. Assigning the result of keyBy to a plain DataStream<Purchase> variable hides KeyedStream-specific methods at compile time. Keep the specialized type when those methods are needed rather than casting it later.

Several important fluent API types are related to DataStream but are not subclasses of it:

These wrapper and builder types make invalid operation sequences harder to express. A WindowedStream, for example, exposes window computations instead of every general DataStream operation, and DataStreamSink marks the end of that output branch.

Runtime Components #

Flink separates job coordination from execution:

• The Dispatcher accepts submitted jobs and starts a JobMaster for each one.
• A JobMaster coordinates one job: it builds the execution graph, requests slots, schedules subtasks, and coordinates checkpoints and recovery.
• The Flink ResourceManager obtains and tracks worker resources from the underlying platform.
• A TaskManager is a Flink worker process. It executes operator subtasks and exchanges records with other TaskManagers.
• A task slot is a TaskManager’s unit of execution capacity. One TaskManager normally has multiple slots.

JobManager is commonly used as an umbrella name for Flink’s control-plane process. Depending on deployment mode, that process can contain the Dispatcher, Flink ResourceManager, REST endpoint, and one or more JobMasters. A Kubernetes TaskManager pod or YARN TaskManager container is the platform resource envelope; the TaskManager is the Flink worker process inside it.

Bounded and Unbounded Streams #

An unbounded stream has no predefined end. A Kafka topic used for live events is normally unbounded, so the Flink job keeps running and processes new records.

A bounded stream has a finite end. Flink can also read a fixed range of Kafka offsets and process it as a batch. The same APIs can therefore support both continuous pipelines and finite reprocessing jobs.

Keyed State #

keyBy partitions records by a logical key, such as customerId. All records for the same key are sent to the same downstream subtask:

events
    .keyBy(Purchase::customerId)
    .process(new CustomerPurchaseFunction());

The process function can maintain state independently for each customer. Flink redistributes that state when the job is restored with different parallelism.

Event Time and Watermarks #

Processing time is the time at which Flink handles a record. Event time is the time at which the event occurred, usually carried in the record itself.

Event time produces more stable results when Kafka records are delayed or arrive out of order. Watermarks tell Flink how far event time is believed to have progressed. A bounded out-of-orderness strategy allows some delay:

WatermarkStrategy<Purchase> watermarks =
        WatermarkStrategy
            .<Purchase>forBoundedOutOfOrderness(Duration.ofSeconds(10))
            .withTimestampAssigner(
                (purchase, previousTimestamp) -> purchase.occurredAt()
            );

When a watermark passes the end of a window, Flink can emit that window’s result. Events that arrive after the watermark may require an allowed-lateness policy or side output.

Checkpoints and Savepoints #

A checkpoint is a consistent snapshot of:

• Kafka source positions;
• operator and keyed state;
• timers; and
• supported sink transaction state.

If a job fails, Flink restores the latest completed checkpoint and resumes from its recorded Kafka offsets.

A savepoint is a user-triggered snapshot intended for operational actions such as an upgrade, migration, or planned rescaling. Checkpoints and savepoints use similar state mechanisms but have different lifecycles and purposes.

Deployment and Component Lifecycles #

Flink does not depend on YARN. It can run on Kubernetes, YARN, a standalone cluster, local JVM processes, or a managed Flink service. The resource manager starts Flink processes; Flink itself schedules operators, manages state, and coordinates checkpoints.

No Flink process is inherently permanent. A streaming job may run indefinitely, but its JobManager, TaskManagers, and subtask execution attempts remain replaceable. The useful distinction is between a logical job, whose state can survive recovery, and the process incarnations currently executing it.

Flink cluster
|
+-- control-plane process
|   +-- Dispatcher                 cluster lifetime
|   +-- Flink ResourceManager      cluster lifetime
|   +-- REST endpoint / blob store cluster lifetime
|   `-- JobMaster                  job lifetime
|       +-- scheduler              job lifetime
|       +-- checkpoint coordinator job lifetime
|       `-- operator coordinators  job lifetime
|
`-- TaskManagers                   dynamically allocated worker lifetime
    `-- task slots                 TaskManager lifetime
        `-- operator subtasks      execution-attempt lifetime

The deployment mode determines the enclosing cluster lifetime:

• In session mode, one long-lived cluster accepts multiple jobs. The Dispatcher and Flink ResourceManager live until the session is stopped; each job gets a JobMaster for only its own lifetime. TaskManagers can be acquired and released as demand changes.
• In application mode, the cluster is scoped to one application, which may submit one or more jobs. The user main method runs in the cluster, and the cluster normally terminates when the application finishes.
• In legacy per-job mode, one cluster is created for one submitted job and ends with it. Modern Flink deployments generally prefer application mode.

┌─────────────┬────────────────────────────────────────────────────────────────────────┐ │ Mode │ JobManager lifetime │ ├─────────────┼────────────────────────────────────────────────────────────────────────┤ │ Session │ Cluster lifetime; survives jobs finishing or being canceled │ ├─────────────┼────────────────────────────────────────────────────────────────────────┤ │ Application │ Application lifetime; usually ends when all application jobs terminate │ ├─────────────┼────────────────────────────────────────────────────────────────────────┤ │ Per-job (LEGACY DON’T USE) │ Single job lifetime │ └─────────────┴────────────────────────────────────────────────────────────────────────┘

If an operator subtask fails, Flink creates a new execution attempt and restores state from a completed checkpoint. If a TaskManager is lost, Flink obtains replacement capacity and reschedules its tasks. Durable checkpoint storage is therefore independent of ephemeral worker storage.

Flink Client and Command-Line Interface #

The Flink client is the submission and control side of a deployment. It reads the Flink configuration, loads the application JAR, builds or submits the job, uploads required artifacts, and communicates with the cluster’s REST endpoint. It can also list, stop, cancel, and take state snapshots of running jobs. Once a detached submission succeeds, the client is not part of the record-processing path: the JobManager coordinates the job and TaskManagers process its records.

• In session mode, the client invokes the application’s main method locally to construct the JobGraph locally, and submit the compiled graph representation it to an existing cluster.
  • main is never invoked again throughout the cluster, but operator logics (i.e. lambdas) runs on each TaskManager.
• In application mode, the client submit user’s job jar to HDFS’s staging dir, request an YARN AM to run the JobManager.
  • main runs in the JobManager process.

The general command form is:

./bin/flink <action> [Flink options] <application.jar> [application arguments]

Options before the JAR configure Flink. Arguments after the JAR are passed to the application’s main method. The exact options depend on the installed Flink version, so the local client is the authoritative reference:

./bin/flink --help
./bin/flink run --help
./bin/flink stop --help
./bin/flink savepoint --help

Submit a Job #

This example submits a Java job to an existing cluster:

./bin/flink run \
  --target remote \
  --jobmanager jobmanager.example.com:8081 \
  --detached \
  --parallelism 8 \
  --class com.example.OrderPipeline \
  order-pipeline.jar \
  --input-topic orders \
  --output-topic customer-totals

Frequently used run options are:

Without --detached, the client remains attached and waits for the job. With --detached, it prints the new JobID and exits after successful submission. The cluster owns the running job, so terminating a detached client does not terminate a job in a session cluster.

-D can set any applicable configuration key. For example:

./bin/flink run \
  -Dpipeline.max-parallelism=128 \
  -Dexecution.checkpointing.interval=30s \
  --detached \
  order-pipeline.jar

When submitting to an existing session cluster, only execution-level configuration can be changed; cluster process settings were fixed when that cluster started.

Select a Deployment Target #

The target tells the client whether to contact an existing cluster or create an application-scoped cluster:

Examples for the main targets are:

# Local MiniCluster
./bin/flink run --target local order-pipeline.jar

# Existing standalone cluster
./bin/flink run \
  --target remote \
  --jobmanager localhost:8081 \
  order-pipeline.jar

# Existing YARN session
./bin/flink run \
  --target yarn-session \
  -Dyarn.application.id=application_123_001 \
  order-pipeline.jar

# New YARN application cluster
./bin/flink run \
  --target yarn-application \
  -Dyarn.application.name=order-pipeline \
  order-pipeline.jar

# Existing Kubernetes session
./bin/flink run \
  --target kubernetes-session \
  -Dkubernetes.cluster-id=flink-session \
  order-pipeline.jar

# New Kubernetes application cluster
./bin/flink run \
  --target kubernetes-application \
  -Dkubernetes.cluster-id=order-pipeline \
  -Dkubernetes.container.image.ref=example/order-pipeline:1.0 \
  local:///opt/flink/usrlib/order-pipeline.jar

For Kubernetes application mode, local:/// identifies a JAR already present inside the container image. Flink can also use its configured artifact management support or a JAR stored in a supported remote filesystem. YARN and Kubernetes require their own platform-specific configuration and credentials in addition to these CLI arguments.

Inspect and Control Jobs #

info builds the application plan without submitting the job:

./bin/flink info \
  --class com.example.OrderPipeline \
  --parallelism 8 \
  order-pipeline.jar

Use list to inspect jobs known to the selected cluster:

./bin/flink list
./bin/flink list --running
./bin/flink list --all

# Select a particular remote cluster.
./bin/flink list \
  --target remote \
  --jobmanager jobmanager.example.com:8081

The JobID printed by run and list identifies the job in lifecycle commands. A savepoint can be triggered explicitly:

./bin/flink savepoint \
  <JOB_ID> \
  s3://flink-state/savepoints

# Return after Flink accepts the savepoint request.
./bin/flink savepoint \
  <JOB_ID> \
  s3://flink-state/savepoints \
  -detached

If execution.checkpointing.savepoint-dir is configured, the target directory can be omitted. A savepoint can also be disposed through Flink so that both its data and metadata are cleaned up:

./bin/flink savepoint \
  --dispose s3://flink-state/savepoints/savepoint-abcd

Recent Flink clients can manually trigger a checkpoint, including a full checkpoint when the configured backend normally creates incremental ones:

./bin/flink checkpoint <JOB_ID>
./bin/flink checkpoint <JOB_ID> --full

Restore a replacement job from a savepoint with --fromSavepoint:

./bin/flink run \
  --detached \
  --fromSavepoint s3://flink-state/savepoints/savepoint-abcd \
  order-pipeline.jar

By default, every state entry in the savepoint must map to an operator in the new job. --allowNonRestoredState permits unmatched state to be discarded, which may be necessary after deliberately removing an operator:

./bin/flink run \
  --fromSavepoint s3://flink-state/savepoints/savepoint-abcd \
  --allowNonRestoredState \
  order-pipeline.jar

Use stop for a controlled streaming-job shutdown that creates a final savepoint:

./bin/flink stop \
  --savepointPath s3://flink-state/savepoints \
  <JOB_ID>

Adding --drain emits a final maximum watermark before the last snapshot so that event-time timers and windows can finish:

./bin/flink stop \
  --drain \
  --savepointPath s3://flink-state/savepoints \
  <JOB_ID>

Draining is appropriate when permanently terminating the pipeline. Avoid it when the job will be resumed later, because firing all event-time timers before the snapshot can change the semantics of the restored pipeline.

Use cancel when the job must be terminated immediately without a final savepoint:

./bin/flink cancel <JOB_ID>

Cluster-selection options also apply to lifecycle commands. For example, a YARN application can be selected with -Dyarn.application.id, and a Kubernetes cluster with -Dkubernetes.cluster-id. This is necessary when the client cannot discover the intended cluster from its local configuration.

How Flink Maps onto YARN #

For a YARN deployment, the component mapping is:

The startup and scheduling sequence is:

1. The Flink client stages artifacts and submits a YARN application.
2. YARN’s ResourceManager selects a NodeManager to launch the ApplicationMaster container.
3. The AM starts the Dispatcher, Flink ResourceManager, REST services, and, when a job is submitted, its JobMaster.
4. The JobMaster determines the slots required by the execution graph.
5. The Flink ResourceManager translates those needs into YARN container requests through the normal ApplicationMaster protocol.
6. YARN NodeManagers launch TaskManager processes in the allocated containers.
7. TaskManagers register, offer task slots, and receive operator subtasks from the JobMaster.

YARN understands applications, queues, nodes, containers, CPU, memory, and process health. It does not understand Flink operators, Kafka partitions, task slots, watermarks, or checkpoints. Flink schedules that work inside the resources YARN supplies.

Failure handling follows the same boundary:

• A task failure is handled by the JobMaster and Flink’s restart strategy.
• A TaskManager or NodeManager loss causes affected tasks to be restored onto replacement capacity.
• An AM failure ends that control-plane process incarnation. YARN can start a new application attempt; full job recovery also requires Flink HA metadata and externally stored checkpoints.
• YARN ResourceManager failover is handled by YARN HA, after which the Flink AM re-registers through the normal YARN protocol.

This resembles Spark on YARN, but Flink exposes an additional control-plane layer. A Spark executor is roughly comparable to a TaskManager, while a Spark driver is closest to a per-job JobMaster. The Dispatcher and Flink ResourceManager manage job admission and worker resources around those JobMasters.

Add the Kafka Connector #

Flink’s Kafka connector is distributed separately from the core runtime. A Maven project needs the connector in addition to the Flink APIs:

<properties>
  <flink.version>YOUR_FLINK_VERSION</flink.version>
  <kafka.connector.version>COMPATIBLE_CONNECTOR_VERSION</kafka.connector.version>
</properties>

<dependencies>
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
  </dependency>

  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
  </dependency>

  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka</artifactId>
    <version>${kafka.connector.version}</version>
  </dependency>
</dependencies>

Use versions listed as compatible in the Flink Kafka connector documentation. Connector releases may have a version lifecycle independent of the Flink runtime. Do not assume that two artifacts are compatible merely because their version strings look similar.

When submitting to a cluster, package the connector with the application or install it according to the cluster’s dependency-management convention. Core Flink dependencies are normally supplied by the cluster and marked provided; connector packaging depends on how that cluster is managed.

Consume a Kafka Topic #

The modern DataStream connector uses KafkaSource. This example reads UTF-8 strings from an orders topic:

import java.time.Duration;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public final class OrderPipeline {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        env.enableCheckpointing(30_000, CheckpointingMode.EXACTLY_ONCE);

        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("kafka:9092")
                .setTopics("orders")
                .setGroupId("order-pipeline")
                .setStartingOffsets(OffsetsInitializer.committedOffsets())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        DataStream<String> orders = env.fromSource(
                source,
                WatermarkStrategy.noWatermarks(),
                "orders-source"
        );

        orders
                .filter(order -> !order.isBlank())
                .name("discard-empty-orders")
                .print();

        env.execute("order-pipeline");
    }
}

The important source settings are:

• bootstrap servers: the initial Kafka brokers used for discovery;
• topics or topic pattern: the input partitions Flink should discover;
• group ID: the Kafka consumer group identity;
• starting offsets: where a job without restored Flink state begins; and
• deserializer: how Kafka records become application objects.

OffsetsInitializer.committedOffsets() requires committed offsets for the partitions. Its overload can specify a reset strategy for partitions without a committed position. Other initializers can start from the earliest offsets, latest offsets, a timestamp, or explicit offsets:

.setStartingOffsets(OffsetsInitializer.earliest())

The starting-offset policy matters only when no usable Flink state is being restored. After recovery from a checkpoint or savepoint, the positions stored in that Flink snapshot take precedence.

Deserialize Keys, Headers, and Metadata #

SimpleStringSchema receives only the record value. Applications that need Kafka keys, headers, timestamps, partition numbers, or offsets should implement KafkaRecordDeserializationSchema<T> instead:

KafkaSource<OrderEvent> source = KafkaSource.<OrderEvent>builder()
        .setBootstrapServers("kafka:9092")
        .setTopics("orders")
        .setGroupId("order-pipeline")
        .setStartingOffsets(OffsetsInitializer.earliest())
        .setDeserializer(new OrderEventDeserializationSchema())
        .build();

Keep deserialization deterministic. Invalid records should follow an explicit policy, such as failing the job or being sent to a dead-letter stream with the original topic, partition, offset, and error. Silently dropping malformed business events makes recovery and auditing difficult.

Discover New Partitions #

Kafka topics can gain partitions while a job is running. The connector can periodically discover them:

.setProperty("partition.discovery.interval.ms", "60000")

New partitions need a defined starting-offset policy. Increasing a topic’s partition count also changes key-to-partition assignment for producers, which can affect ordering assumptions outside Flink.

Process Events with Event Time #

Suppose each Kafka value is parsed into this record:

public record Purchase(
        String customerId,
        long amountInCents,
        long occurredAt
) {
}

Flink can compute one-minute totals per customer:

import java.time.Duration;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

WatermarkStrategy<Purchase> watermarks =
        WatermarkStrategy
            .<Purchase>forBoundedOutOfOrderness(Duration.ofSeconds(10))
            .withTimestampAssigner(
                (purchase, previousTimestamp) -> purchase.occurredAt()
            )
            .withIdleness(Duration.ofMinutes(1));

DataStream<Purchase> purchases = rawOrders
        .map(OrderPipeline::parsePurchase)
        .assignTimestampsAndWatermarks(watermarks);

DataStream<Purchase> minuteTotals = purchases
        .keyBy(Purchase::customerId)
        .window(TumblingEventTimeWindows.of(Time.minutes(1)))
        .reduce((left, right) -> new Purchase(
                left.customerId(),
                left.amountInCents() + right.amountInCents(),
                Math.max(left.occurredAt(), right.occurredAt())
        ));

withIdleness prevents an idle Kafka partition from holding back the combined watermark indefinitely. Choose its duration from observed traffic patterns, not arbitrarily: a value that is too short can make a temporarily quiet partition appear complete too early.

Kafka preserves order within one partition, not across a topic. Flink’s event time model handles cross-partition interleaving, but the application must still define how much lateness is acceptable and what to do with very late events.

Publish Results to Kafka #

KafkaSink writes a DataStream to Kafka:

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;

KafkaSink<String> sink = KafkaSink.<String>builder()
        .setBootstrapServers("kafka:9092")
        .setRecordSerializer(
            KafkaRecordSerializationSchema.builder()
                .setTopic("order-totals")
                .setValueSerializationSchema(new SimpleStringSchema())
                .build()
        )
        .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
        .build();

results.sinkTo(sink).name("order-totals-sink");

Use KafkaRecordSerializationSchema<T> directly when the output topic, key, headers, or timestamp depends on the record.

The sink supports three delivery guarantees:

• NONE: does not wait for Kafka acknowledgements as part of checkpointing;
• AT_LEAST_ONCE: ensures acknowledged records are not lost, but recovery may produce duplicates; and
• EXACTLY_ONCE: uses Kafka transactions coordinated with Flink checkpoints.

Exactly-once output requires checkpointing:

KafkaSink<String> sink = KafkaSink.<String>builder()
        .setBootstrapServers("kafka:9092")
        .setRecordSerializer(
            KafkaRecordSerializationSchema.builder()
                .setTopic("order-totals")
                .setValueSerializationSchema(new SimpleStringSchema())
                .build()
        )
        .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
        .setTransactionalIdPrefix("order-pipeline-")
        .build();

Every concurrently running job needs a unique transactional ID prefix. Reusing a prefix can cause one producer to fence another. Kafka consumers that must not observe uncommitted transactional output should use isolation.level=read_committed.

Exactly-once is an end-to-end property, not merely a sink option. It requires:

• replayable input;
• deterministic or correctly checkpointed processing;
• successful Flink checkpoints;
• a transactional or idempotent destination; and
• consumers that observe committed results appropriately.

An external side effect inside a map or process function, such as sending an HTTP request, is not automatically rolled back when Flink restores a checkpoint. Use an idempotency key, a transactional integration, or a design that publishes an intent to Kafka for a separate system to execute.

How Offsets and Checkpoints Work Together #

Kafka consumer groups usually track progress through committed offsets. Flink also stores source positions in its checkpoints. These mechanisms are related but serve different roles:

1. The source reads records from assigned Kafka partitions.
2. A Flink checkpoint barrier moves through the job.
3. Operators snapshot their state consistently with the source positions.
4. The checkpoint completes after all required snapshots and sink actions succeed.
5. On failure, Flink restores operator state and Kafka positions from that completed checkpoint.

Kafka committed offsets are useful for visibility and for an initial start policy, but they are not a replacement for Flink checkpoint state. Deleting Flink state and restarting from consumer-group offsets can lose the exact relationship between source progress and operator state.

Checkpoint storage must be durable and accessible to the cluster. A local filesystem path on an ephemeral worker is not sufficient for production. Retain checkpoints or take savepoints according to the recovery and deployment strategy.

Use Kafka with Flink SQL #

Flink’s Table API and SQL connector expose a Kafka topic as a dynamic table. The following table reads JSON order events:

CREATE TABLE orders (
  order_id STRING,
  customer_id STRING,
  amount DECIMAL(12, 2),
  occurred_at TIMESTAMP(3),
  WATERMARK FOR occurred_at AS occurred_at - INTERVAL '10' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'orders',
  'properties.bootstrap.servers' = 'kafka:9092',
  'properties.group.id' = 'order-sql',
  'scan.startup.mode' = 'group-offsets',
  'format' = 'json',
  'json.fail-on-missing-field' = 'false',
  'json.ignore-parse-errors' = 'false'
);

A second Kafka table can receive aggregates:

CREATE TABLE customer_totals (
  customer_id STRING,
  window_start TIMESTAMP(3),
  window_end TIMESTAMP(3),
  total_amount DECIMAL(18, 2),
  PRIMARY KEY (customer_id, window_start) NOT ENFORCED
) WITH (
  'connector' = 'upsert-kafka',
  'topic' = 'customer-totals',
  'properties.bootstrap.servers' = 'kafka:9092',
  'key.format' = 'json',
  'value.format' = 'json'
);

The query groups events into event-time windows:

INSERT INTO customer_totals
SELECT
  customer_id,
  window_start,
  window_end,
  SUM(amount)
FROM TABLE(
  TUMBLE(TABLE orders, DESCRIPTOR(occurred_at), INTERVAL '1' MINUTE)
)
GROUP BY customer_id, window_start, window_end;

Use the regular kafka connector for append-only records. Use upsert-kafka when the table produces updates or deletes identified by a primary key. An ordinary Kafka topic does not magically become an upsert table: the key format, value format, changelog semantics, and downstream consumer must agree.

SQL connector artifacts must also be available to the Flink cluster. The exact installation mechanism differs between a SQL client, an application JAR, and a managed Flink service.

Partitioning and Parallelism #

Kafka partitions determine the maximum number of source subtasks that can read one topic concurrently. With 12 partitions and source parallelism 20, at least eight source subtasks have no partition to read. With source parallelism 4, each subtask reads multiple partitions.

TaskManager count is independent of Kafka partition count. The relevant limit is Kafka source parallelism, not worker-process count. For example, all of the following can provide 12 source slots for a 12-partition topic:

3 TaskManagers x 4 slots
6 TaskManagers x 2 slots
12 TaskManagers x 1 slot

A job may also have more TaskManagers or downstream parallelism than source partitions because parsing, asynchronous API calls, keyed processing, or sinks can require more capacity than reading Kafka. Increasing source parallelism beyond the discovered partition count leaves some source subtasks idle, but does not impose a corresponding upper bound on the rest of the job.

That does not mean every operator should have the same parallelism. After keyBy, Flink repartitions records according to the logical key and can run downstream operators at a different parallelism.

Choose partition counts and parallelism using:

• peak input and output throughput;
• record size and serialization cost;
• skew in Kafka keys and Flink keys;
• state size per key;
• checkpoint duration;
• broker and network capacity; and
• the desired rescaling range.

More partitions create more potential concurrency, but also more files, connections, metadata, and checkpoint work. They do not fix a hot key: all records for one Flink key still go to one keyed subtask unless the application changes its aggregation strategy.

Schema Management #

JSON is convenient for examples but does not by itself provide strong schema compatibility. Production pipelines commonly use Avro, Protobuf, or JSON Schema with a schema registry.

Plan for schema evolution:

• assign stable field names and meanings;
• distinguish required fields from optional fields;
• use compatible defaults when adding fields;
• avoid reusing a field for a new meaning;
• preserve unknown fields when consumers may need them; and
• test old records against new application versions.

Kafka retention makes schema compatibility especially important. A restored or reprocessing job may read events produced months earlier.

Deserialization failures should carry enough context to diagnose the source record. If the organization permits dead-letter topics, include the original topic, partition, offset, key, schema identifier, and a safe error description. Avoid copying sensitive payloads into unrestricted logs.

Security Configuration #

The connector accepts Kafka client properties for TLS and SASL. A deployment might supply properties such as:

.setProperty("security.protocol", "SASL_SSL")
.setProperty("sasl.mechanism", "SCRAM-SHA-512")
.setProperty(
    "sasl.jaas.config",
    System.getenv("KAFKA_SASL_JAAS_CONFIG")
)

Do not hard-code passwords or JAAS credentials in source code. Inject them from the platform’s secret mechanism and restrict access to job configuration and logs.

Apply least privilege to Kafka ACLs. A source job normally needs topic description and read access plus permission for its consumer group. A sink needs write access and, for exactly-once delivery, the permissions required for transactional producers.

Use encrypted broker connections, validate broker certificates, and avoid disabling hostname verification as a shortcut.

Operating the Pipeline #

A healthy Kafka-Flink pipeline needs visibility into both systems.

Monitor at least:

• input and output records and bytes;
• Kafka consumer lag per partition;
• busy, idle, and backpressured time for Flink subtasks;
• checkpoint duration, size, failures, and alignment time;
• restart counts and failure causes;
• watermark progression and idle partitions;
• late or invalid record counts;
• state size and growth;
• Kafka produce latency and errors; and
• transaction timeouts for exactly-once sinks.

Consumer lag alone is not enough. A job can have low lag while producing incorrect results, repeatedly failing checkpoints, or accumulating state without bounds. Conversely, a deliberate bounded replay can have high lag while operating correctly.

Set stable operator UIDs for stateful production jobs:

DataStream<OrderTotal> totals = purchases
        .keyBy(Purchase::customerId)
        .process(new CustomerTotalFunction())
        .uid("customer-total")
        .name("customer-total");

Flink uses operator identity to map saved state during upgrades. Explicit UIDs reduce the risk that a harmless code-graph change prevents state restoration. Do not change or reuse them casually.

Test recovery before production. A useful test kills a task while records are flowing, waits for restoration, and verifies the final output for duplicates, loss, and ordering assumptions. Also test a savepoint upgrade, rescaling, Kafka broker interruption, malformed records, and a slow destination.

Common Pitfalls #

Assuming Kafka Ordering Is Global #

Kafka orders records only within a partition. If all events for one entity must remain ordered, producers should use a stable key so those events reach the same partition. A later Flink keyBy can regroup events but cannot recreate a global order that never existed.

Confusing Processing Guarantees #

EXACTLY_ONCE checkpoint mode does not make arbitrary external calls exactly-once. Trace every side effect through failure and replay behavior.

Starting from the Wrong Offsets #

earliest, latest, committed offsets, timestamps, and restored state have different meanings. Make restart and full-replay procedures explicit before an incident occurs.

Ignoring Watermark Idleness #

One quiet Kafka partition can stop a downstream event-time watermark. Configure idleness when partitions may legitimately receive no records.

Using an Unbounded Key or State Model #

A key such as a request ID can create millions of tiny state entries. State needs a cleanup mechanism: windows, timers, state time-to-live, or another business rule with understood correctness implications.

Treating a Dead-Letter Topic as a Fix #

A dead-letter topic preserves problematic records; it does not resolve them. Define ownership, alerting, retention, replay, and privacy controls around it.

A Practical Design Checklist #

Before deploying a Flink job connected to Kafka, answer:

1. Which Kafka key preserves the required ordering?
2. Where does a new deployment start when no Flink state exists?
3. Which event timestamp drives the computation?
4. How late may events arrive, and what happens after that limit?
5. What state does each key retain, and how is it cleaned up?
6. Which delivery guarantee is required at every output?
7. Are all external side effects replay-safe?
8. Where are checkpoints stored, and how long are they retained?
9. How will the job be upgraded, rolled back, and rescaled?
10. How are schemas evolved and malformed records handled?
11. Which metrics and alerts reveal lag, backpressure, and checkpoint failure?
12. Have failure and recovery been tested with realistic traffic?

Kafka gives Flink a durable, replayable event log. Flink adds distributed state, event-time processing, recovery, and rich transformations. Used together, they form a strong foundation for streaming applications, but correctness depends on deliberate choices about offsets, checkpoints, transactions, schemas, partitioning, and every side effect beyond the pipeline.