Is anyone familiar with a best-practices example for doing exactly-once processing as it pertains to Kafka in Flink? I’ve been continually encountering errors surrounding it enough to make me feel like there’s something wrong or misconfigured in my job(s). It’d be nice to have some type of official example of one that’s working as expected to compare against to rule out any misconfiguration.

what errors did you encounter? I was trying to use

transaction api

for my app as well, but was very successful. My issue was that the Flink job couldn’t recover from the savepoint after an upgrade.

I’ve been continually encountering errors, specifically around deployments of the job (i.e. suspending with savepoint and restoration from the same savepoint with the updated jar). It typically ends up being an

InvalidPidMappingException

related to the transactions in Kafka which puts the job into a state where some manual intervention is required. The sources and sinks aren’t terribly crazy either:

// Example Source
KafkaSource.builder<T>()
    .setBootstrapServers(parameterTool.get("bootstrap-servers"))
    .setTopics(topic)
    .setGroupId(parameterTool.get("consumer-group-id"))
    .setProperty("isolation.level", "read_committed")
    .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
    .setDeserializer(...)

// Example Sink
val producerProperties = Properties().apply {
    setProperty("<http://transaction.timeout.ms|transaction.timeout.ms>", parameterTool.get("transaction-timeout-ms"))
}

val transactionPrefix = "${parameterTool.get("transaction-prefix")}-${parameterTool.get("version")}-$topic"
KafkaSink.builder<T>()
    .setBootstrapServers(bootstrapServers)
    .setDeliverGuarantee(deliveryGuarantee)       // DeliveryGuarantee.EXACTLY_ONCE
    .setKafkaProducerConfig(producerProperties)
    .setTransactionalIdPrefix(transactionPrefix) 
    .setRecordSerializer(...)
    .build()

Cool. Here is the previous thread about Kafka transaction issues I had. https://apache-flink.slack.com/archives/C03G7LJTS2G/p1684968972410419