Kafka Transaction Producer Sequence Number Conflicts with Idempotent Producers

[alexisvisco]

Hello folks, we have a problem using transaction to ensure batch messages are either all sent or not sent.

We have a Kafka producer setup that:

• Uses idempotent producers (ensures messages are sent exactly once)
• Creates unique transaction IDs for each producer using KSUID
• Uses a producer pool to manage these producers
• Has multiple application replicas running

We're seeing this error in Kafka logs:

2024-11-13 12:42:18 | 2024-11-13 11:42:18,874 ERROR [ReplicaManager broker=0] Error processing append operation on partition apiserver-jobs-53 (kafka.server.ReplicaManager) [data-plane-kafka-request-handler-4]
-- | --
2024-11-13 12:42:18 | org.apache.kafka.common.errors.OutOfOrderSequenceException: Out of order sequence number for producer 2050 at offset 10021 in partition apiserver-jobs-53: 123 (incoming seq. number), 58 (current end sequence number)

What's puzzling:

• Since each producer gets a unique transaction ID, they should have independent sequence tracking
• But somehow we're seeing sequence number conflicts for the same producer ID (2050)
• This only happens after some time running, and fixes itself on restart and then re-appear

I don't know how to fix that because I heavily inspired myself from the example of the txn producer, but I have sync producer instead of async producer.

Here are the configurations of my producer :

producerProvider := newProducerProvider(brokers, func() *sarama.Config {
		config := sarama.NewConfig()
		config.Producer.Idempotent = true
		config.Producer.Return.Successes = true
		config.Producer.RequiredAcks = sarama.WaitForAll
		config.Producer.Partitioner = sarama.NewHashPartitioner
		config.Producer.Transaction.Retry.Backoff = time.Millisecond * 100
		config.Producer.Transaction.Retry.Max = 512
		config.Net.MaxOpenRequests = 1
		return config
	})

func newProducerProvider(brokers []string, producerConfigurationProvider func() *sarama.Config) *producerPool {
	provider := &producerPool{}
	provider.producerProvider = func() (sarama.SyncProducer, error) {
		config := producerConfigurationProvider()
		config.Producer.Transaction.ID = ksuid.New().String()
		producer, err := sarama.NewSyncProducer(brokers, config)
		if err != nil {
			return nil, fmt.Errorf("failed to create producer: %w", err)
		}
		return producer, nil
	}
	return provider
}

I am using the latest version of sarama with kafka 3.8.0.

And by the way I don't understand why AsyncProducer is go to in example with transactions, because having a transaction required to wait to know each messages are sent ? Isn't contradictory ?