Function TestBrokerProducerFlushSkipsMutedPartitions

async_producer_test.go:1862–1900

(t *testing.T)

}

func TestBrokerProducerFlushSkipsMutedPartitions(t *testing.T) {
	config := NewTestConfig()
	parent := &asyncProducer{
		conf:   config,
		muter:  newPartitionMuter(),
		txnmgr: &transactionManager{},
	}
	bp := &brokerProducer{
		parent:            parent,
		accumulatingBatch: newProduceSet(parent),
		currentRetries:    make(map[string]map[int32]error),
	}

	safeAddMessage(t, bp.accumulatingBatch, &ProducerMessage{Topic: "topic", Partition: 0, Value: StringEncoder("p0")})
	safeAddMessage(t, bp.accumulatingBatch, &ProducerMessage{Topic: "topic", Partition: 1, Value: StringEncoder("p1")})

	blocked := newProduceSet(parent)
	safeAddMessage(t, blocked, &ProducerMessage{Topic: "topic", Partition: 1, Value: StringEncoder("held")})
	if !parent.muter.tryMute(blocked) {
		t.Fatal("expected to mute blocked partition")
	}
	defer parent.muter.unmute(blocked)

	if !bp.tryBuildFlushingBatch() {
		t.Fatal("expected to flush available partitions")
	}
	if bp.flushingBatch == nil {
		t.Fatal("expected flushing batch to be set")
	}
	if _, ok := bp.flushingBatch.msgs["topic"][1]; ok {
		t.Fatal("expected muted partition to stay buffered")
	}
	if _, ok := bp.accumulatingBatch.msgs["topic"][0]; ok {
		t.Fatal("expected unmuted partition to flush")
	}
	if _, ok := bp.accumulatingBatch.msgs["topic"][1]; !ok {
		t.Fatal("expected muted partition to remain in accumulating batch")
	}
}

// TestBrokerProducerWaitForSpaceAllPartitionsMuted verifies that waitForSpace unblocks
// when all partitions in the accumulating batch are externally muted and later unmuted.