This version is still in development and is not considered stable yet. For the latest stable version, please use Korvet 0.19!

Apache Flink Integration

This guide covers integrating Korvet with Apache Flink using the standard Flink Kafka connector.

Overview

Flink connects to Korvet through the Flink Kafka connector, with no Korvet-specific configuration. This allows you to:

  • Stream data from Korvet topics into Flink jobs with KafkaSource

  • Write processed data back to Korvet topics with KafkaSink

  • Run bounded (batch-style) reads over a topic, or start from a timestamp

Supported delivery guarantees:

Guarantee Status Notes

NONE

✅ Supported

Fire-and-forget writes

AT_LEAST_ONCE

✅ Supported

Default guarantee; works with the idempotent producer enabled by default in Kafka clients 3.x

EXACTLY_ONCE

❌ Not supported

Requires Kafka transactions, which Korvet does not implement. Jobs configured for exactly-once fail at startup instead of silently downgrading. See Kafka Compatibility.

Prerequisites

  • Korvet server running and reachable from the Flink cluster on port 9092

  • The flink-connector-kafka dependency on your job’s classpath

  • A Kafka topic created in Korvet

Reading from Korvet

KafkaSource works against Korvet with standard options:

KafkaSource<String> source = KafkaSource.<String>builder()
        .setBootstrapServers("<korvet-host>:9092")
        .setTopics("my-topic")
        .setGroupId("my-flink-group")
        .setStartingOffsets(OffsetsInitializer.earliest())
        .setValueOnlyDeserializer(new SimpleStringSchema())
        .build();

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

Starting From a Timestamp

To avoid replaying an entire topic, start from a time boundary:

KafkaSource<String> source = KafkaSource.<String>builder()
        .setBootstrapServers("<korvet-host>:9092")
        .setTopics("my-topic")
        .setGroupId("my-flink-group")
        .setStartingOffsets(OffsetsInitializer.timestamp(1711324800000L))
        .setValueOnlyDeserializer(new SimpleStringSchema())
        .build();

Bounded Reads

For batch-style processing over the current contents of a topic:

KafkaSource<String> source = KafkaSource.<String>builder()
        .setBootstrapServers("<korvet-host>:9092")
        .setTopics("my-topic")
        .setGroupId("my-flink-group")
        .setStartingOffsets(OffsetsInitializer.earliest())
        .setBounded(OffsetsInitializer.latest())
        .setValueOnlyDeserializer(new SimpleStringSchema())
        .build();

Offsets and Checkpoints

Flink stores source offsets in its own checkpoints, so exactly-once processing within Flink works normally when reading from Korvet. When checkpointing is enabled, the source also commits offsets to Korvet’s consumer group storage, which makes progress visible to monitoring tools and kafka-consumer-groups.

Writing to Korvet

Use KafkaSink with AT_LEAST_ONCE (or NONE):

KafkaSink<String> sink = KafkaSink.<String>builder()
        .setBootstrapServers("<korvet-host>:9092")
        .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
        .setRecordSerializer(KafkaRecordSerializationSchema.builder()
                .setTopic("output-topic")
                .setValueSerializationSchema(new SimpleStringSchema())
                .build())
        .build();

stream.sinkTo(sink);

DeliveryGuarantee.EXACTLY_ONCE requires Kafka transactions and does not work with Korvet. The transactional producer fails during initialization (InitProducerId is rejected for transactional IDs), so the job fails at startup. Use AT_LEAST_ONCE with idempotent downstream processing instead.

Troubleshooting

Timeout Connecting to the Broker

If the job fails with a timeout waiting for a node assignment, Flink cannot reach Korvet. Check network connectivity to port 9092 and, when Korvet runs behind a load balancer or in a container, configure the advertised address:

export KORVET_BROKER_ADVERTISED_HOST=<reachable-hostname>
export KORVET_BROKER_ADVERTISED_PORT=9092

Exactly-Once Job Fails at Startup

A KafkaSink configured with EXACTLY_ONCE and a transactionalIdPrefix fails when the producer initializes. This is expected: Korvet does not support Kafka transactions. Switch the sink to AT_LEAST_ONCE.

Next Steps