Function TestSyncProducerToNonExistingTopic

sync_producer_test.go:255–285

(t *testing.T)

}

func TestSyncProducerToNonExistingTopic(t *testing.T) {
	broker := NewMockBroker(t, 1)

	metadataResponse := new(MetadataResponse)
	metadataResponse.AddBroker(broker.Addr(), broker.BrokerID())
	metadataResponse.AddTopicPartition("my_topic", 0, broker.BrokerID(), nil, nil, nil, ErrNoError)
	broker.Returns(metadataResponse)

	config := NewTestConfig()
	config.Metadata.Retry.Max = 0
	config.Producer.Retry.Max = 0
	config.Producer.Return.Successes = true

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

	metadataResponse = new(MetadataResponse)
	metadataResponse.AddBroker(broker.Addr(), broker.BrokerID())
	metadataResponse.AddTopic("unknown", ErrUnknownTopicOrPartition)
	broker.Returns(metadataResponse)

	_, _, err = producer.SendMessage(&ProducerMessage{Topic: "unknown"})
	if !errors.Is(err, ErrUnknownTopicOrPartition) {
		t.Error("Uxpected ErrUnknownTopicOrPartition, found:", err)
	}

	safeClose(t, producer)
	broker.Close()
}

func TestSyncProducerRecoveryWithRetriesDisabled(t *testing.T) {
	seedBroker := NewMockBroker(t, 1)