Function TestAsyncProducerPartitionUnmuting

async_producer_test.go:3054–3194

TestAsyncProducerPartitionUnmuting verifies that partitions are properly unmuted in all error paths: send errors, NoResponse acks, etc. Without proper unmuting, partitions remain muted and subsequent messages would block indefinitely.

(t *testing.T)

// in all error paths: send errors, NoResponse acks, etc. Without proper unmuting,
// partitions remain muted and subsequent messages would block indefinitely.
func TestAsyncProducerPartitionUnmuting(t *testing.T) {
	const topic = "test_topic"

	t.Run("NoResponse acks unmute partitions", func(t *testing.T) {
		broker := NewMockBroker(t, 1)
		defer broker.Close()

		metadataResponse := NewMockMetadataResponse(t).
			SetBroker(broker.Addr(), broker.BrokerID()).
			SetLeader(topic, 0, broker.BrokerID())
		broker.SetHandlerByMap(map[string]MockResponse{
			"MetadataRequest": metadataResponse,
		})

		config := NewTestConfig()
		config.Producer.RequiredAcks = NoResponse
		config.Producer.Return.Successes = true
		config.Producer.Flush.Messages = 1
		config.Net.MaxOpenRequests = 5

		producer, err := NewAsyncProducer([]string{broker.Addr()}, config)
		if err != nil {
			t.Fatal(err)
		}

		for range 3 {
			producer.Input() <- &ProducerMessage{
				Topic:     topic,
				Partition: 0,
				Value:     StringEncoder("msg"),
			}
		}

		successCount := 0
		for i := range 3 {
			select {
			case <-producer.Successes():
				successCount++
			case err := <-producer.Errors():
				t.Fatalf("unexpected error: %v", err)
			case <-time.After(5 * time.Second):
				t.Fatalf("timeout waiting for success %d (got %d) - partitions may not be unmuted", i+1, successCount)
			}
		}

		if successCount != 3 {
			t.Errorf("expected 3 successes, got %d", successCount)
		}

		closeProducer(t, producer)
	})

	t.Run("retry keeps partition muted until queued", func(t *testing.T) {
		broker := NewMockBroker(t, 1)
		defer broker.Close()

		metadataResponse := new(MetadataResponse)
		metadataResponse.AddBroker(broker.Addr(), broker.BrokerID())