fix: assign sequence when flushing retry buffers (#3362)

As noted in #3354, messages buffered due to the highWatermark would
inadvertently skip the usual sequence number assignment, manifesting as
OutOfOrderSequenceException on the broker.

Wrote a functional test to trigger this behaviour and then used the same
logic as found in partitionProducer.dispatch() to prevent the problem
occurring.

Signed-off-by: Dominic Evans <[email protected]>

Author: dnwe

async_producer.go

@@ -751,6 +751,10 @@ func (pp *partitionProducer) flushRetryBuffers() {
 		}
 
 		for _, msg := range pp.retryState[pp.highWatermark].buf {
+			if pp.parent.conf.Producer.Idempotent && msg.retries == 0 && msg.flags == 0 && !msg.hasSequence {
+				msg.sequenceNumber, msg.producerEpoch = pp.parent.txnmgr.getAndIncrementSequenceNumber(msg.Topic, msg.Partition)
+				msg.hasSequence = true
+			}
 			pp.brokerProducer.input <- msg
 		}
 

async_producer_test.go

@@ -61,6 +61,49 @@ func expectResults(t *testing.T, p AsyncProducer, successCount, errorCount int)
 	expectResultsWithTimeout(t, p, successCount, errorCount, 5*time.Minute)
 }
 
+func TestPartitionProducerFlushRetryBuffersAssignsSequence(t *testing.T) {
+	cfg := NewTestConfig()
+	cfg.Producer.Idempotent = true
+
+	txnmgr := &transactionManager{
+		producerID:      1,
+		producerEpoch:   0,
+		sequenceNumbers: map[string]int32{"topic-0": 1},
+	}
+
+	parent := &asyncProducer{
+		conf:   cfg,
+		txnmgr: txnmgr,
+	}
+
+	bp := &brokerProducer{
+		input: make(chan *ProducerMessage, 1),
+	}
+
+	pp := &partitionProducer{
+		parent:         parent,
+		topic:          "topic",
+		partition:      0,
+		brokerProducer: bp,
+		retryState:     make([]partitionRetryState, 1),
+		highWatermark:  1,
+	}
+
+	msg := &ProducerMessage{Topic: "topic", Partition: 0}
+	pp.retryState[0].buf = []*ProducerMessage{msg}
+
+	pp.flushRetryBuffers()
+
+	select {
+	case flushed := <-bp.input:
+		require.True(t, flushed.hasSequence, "message should have a sequence assigned")
+		require.Equal(t, int32(1), flushed.sequenceNumber, "sequence number should have increased")
+		require.Equal(t, txnmgr.producerEpoch, flushed.producerEpoch, "producer epoch should be the same")
+	default:
+		t.Fatal("expected buffered message to flush")
+	}
+}
+
 type testPartitioner chan *int32
 
 func (p testPartitioner) Partition(msg *ProducerMessage, numPartitions int32) (int32, error) {

functional_producer_test.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"os/exec"
 	"strconv"
 	"strings"
 	"sync"
@@ -726,6 +727,110 @@ func TestFuncProducingIdempotentWithBrokerFailure(t *testing.T) {
 	}
 }
 
+func TestFuncIdempotentBufferedSequence(t *testing.T) {
+	checkKafkaVersion(t, "0.11.0.0")
+	setupFunctionalTest(t)
+	defer teardownFunctionalTest(t)
+
+	const (
+		topic           = "test.1"
+		partition int32 = 0
+	)
+
+	cfg := NewFunctionalTestConfig()
+	cfg.Net.MaxOpenRequests = 1
+	cfg.Producer.Idempotent = true
+	cfg.Producer.RequiredAcks = WaitForAll
+	cfg.Producer.Return.Successes = true
+	cfg.Producer.Return.Errors = true
+	cfg.Producer.Retry.Max = 64
+	cfg.Producer.Retry.Backoff = 250 * time.Millisecond
+
+	start := time.Now()
+
+	producer, err := NewAsyncProducer(FunctionalTestEnv.KafkaBrokerAddrs, cfg)
+	require.NoError(t, err)
+	defer producer.Close()
+
+	asyncProd, ok := producer.(*asyncProducer)
+	require.True(t, ok)
+
+	waitForMessages := func(count int) {
+		timeout := time.After(2 * time.Minute)
+		for count > 0 {
+			select {
+			case <-timeout:
+				t.Fatalf("timed out waiting for %d messages", count)
+			case perr := <-producer.Errors():
+				if perr != nil {
+					t.Logf("producer error: %v", perr.Err)
+				}
+				count--
+			case <-producer.Successes():
+				count--
+			}
+		}
+	}
+
+	for i := 0; i < 5; i++ {
+		producer.Input() <- &ProducerMessage{
+			Topic:     topic,
+			Partition: partition,
+			Value:     StringEncoder(fmt.Sprintf("warmup-%d", i)),
+		}
+	}
+	waitForMessages(5)
+
+	leader, err := asyncProd.client.Leader(topic, partition)
+	require.NoError(t, err)
+
+	bp := asyncProd.getBrokerProducer(leader)
+	defer asyncProd.unrefBrokerProducer(leader, bp)
+
+	asyncProd.inFlight.Add(1)
+	pp := &partitionProducer{
+		parent:         asyncProd,
+		topic:          topic,
+		partition:      partition,
+		brokerProducer: bp,
+		leader:         leader,
+		retryState:     make([]partitionRetryState, asyncProd.conf.Producer.Retry.Max+1),
+		highWatermark:  1,
+	}
+	pp.retryState[0].buf = []*ProducerMessage{{
+		Topic:     topic,
+		Partition: partition,
+		Value:     StringEncoder("buffered"),
+	}}
+	pp.flushRetryBuffers()
+
+	waitForMessages(1)
+
+	producer.Input() <- &ProducerMessage{
+		Topic:     topic,
+		Partition: partition,
+		Value:     StringEncoder("post-buffer"),
+	}
+	waitForMessages(1)
+
+	logSince := start.UTC().Format(time.RFC3339)
+	cmd := exec.Command(
+		"docker",
+		"compose",
+		"logs",
+		"--since",
+		logSince,
+		fmt.Sprintf("kafka-%d", leader.ID()),
+	)
+	cmd.Env = os.Environ()
+	out, err := cmd.CombinedOutput()
+	require.NoErrorf(t, err, "failed to read broker logs: %s", out)
+
+	logs := string(out)
+	t.Logf("kafka-%d logs since %s:\n%s", leader.ID(), logSince, logs)
+	require.NotContains(t, logs, "OutOfOrderSequenceException", "leader logs contained out-of-order sequence errors:\n%s", logs)
+}
+
 func TestInterceptors(t *testing.T) {
 	config := NewFunctionalTestConfig()
 	setupFunctionalTest(t)