Function TestProducerWithBrokenPartitioner

mocks/async_producer_test.go:211–238

(t *testing.T)

}

func TestProducerWithBrokenPartitioner(t *testing.T) {
	trm := newTestReporterMock()
	config := NewTestConfig()
	config.Producer.Partitioner = func(string) sarama.Partitioner {
		return brokePartitioner{}
	}
	mp := NewAsyncProducer(trm, config)
	mp.ExpectInputWithMessageCheckerFunctionAndSucceed(func(msg *sarama.ProducerMessage) error {
		if msg.Partition != 15 {
			t.Error("Expected partition 15, found: ", msg.Partition)
		}
		if msg.Topic != "test" {
			t.Errorf(`Expected topic "test", found: %q`, msg.Topic)
		}
		return nil
	})
	mp.ExpectInputAndSucceed() // should actually fail in partitioning

	mp.Input() <- &sarama.ProducerMessage{Topic: "test"}
	mp.Input() <- &sarama.ProducerMessage{Topic: "not-test"}
	if err := mp.Close(); err != nil {
		t.Error(err)
	}

	if len(trm.errors) != 1 || !strings.Contains(trm.errors[0], "partitioning unavailable") {
		t.Error("Expected to report partitioning unavailable, found", trm.errors)
	}
}

// brokeProducer refuses to partition anything not on the “test” topic, and sends everything on
// that topic to partition 15.