Function TestJavaMurmur2PartitionerInterop

functional_java_interop_test.go:400–477

TestJavaMurmur2PartitionerInterop verifies that Sarama's NewMurmur2Partitioner routes keyed messages to the same partitions as the Apache Kafka Java client's DefaultPartitioner (which uses murmur2 internally).

(t *testing.T)

// routes keyed messages to the same partitions as the Apache Kafka Java client's
// DefaultPartitioner (which uses murmur2 internally).
func TestJavaMurmur2PartitionerInterop(t *testing.T) {
	setupFunctionalTest(t)
	defer teardownFunctionalTest(t)

	checkKafkaVersion(t, "0.10.0")

	const topic = "test.64"
	const numPartitions = int32(64)

	keyedMessages := []struct{ key, value string }{
		{"foo", "value-foo"},
		{"bar", "value-bar"},
		{"baz", "value-baz"},
		{"kafka", "value-kafka"},
		{"sarama", "value-sarama"},
		{"hello", "value-hello"},
		{"world", "value-world"},
	}

	// record the end offsets per partition before we produce anything
	startOffsets := make([]int64, numPartitions)
	for p := int32(0); p < numPartitions; p++ {
		startOffsets[p] = endOffsetForPartition(t, topic, p)
	}

	// produce with Java's DefaultPartitioner (murmur2)
	produceKeyedWithJava(t, topic, keyedMessages)

	// determine which partition each key landed on via Java
	javaPartition := make(map[string]int32, len(keyedMessages))
	for p := int32(0); p < numPartitions; p++ {
		endOff := endOffsetForPartition(t, topic, p)
		if endOff <= startOffsets[p] {
			continue
		}
		msgs := consumeKeyedFromPartition(t, topic, p, startOffsets[p], int(endOff-startOffsets[p]))
		for key := range msgs {
			javaPartition[key] = p
		}
	}
	require.Len(t, javaPartition, len(keyedMessages), "Java producer did not produce all messages")

	// record new end offsets before Sarama production
	for p := int32(0); p < numPartitions; p++ {
		startOffsets[p] = endOffsetForPartition(t, topic, p)
	}

	// produce the same keyed messages with Sarama's murmur2 partitioner
	config := NewFunctionalTestConfig()
	config.Producer.Partitioner = NewMurmur2Partitioner
	config.Producer.Return.Successes = true
	config.Producer.RequiredAcks = WaitForAll

	producer, err := NewSyncProducer(FunctionalTestEnv.KafkaBrokerAddrs, config)
	require.NoError(t, err)
	defer producer.Close()

	saramaPartition := make(map[string]int32, len(keyedMessages))