Function consumeKeyedFromPartition

functional_java_interop_test.go:368–395

consumeKeyedFromPartition consumes up to count messages from a specific partition, returning them as key+value pairs.

(t *testing.T, topic string, partition int32, startOffset int64, count int)

// consumeKeyedFromPartition consumes up to count messages from a specific partition,
// returning them as key+value pairs.
func consumeKeyedFromPartition(t *testing.T, topic string, partition int32, startOffset int64, count int) map[string]string {
	t.Helper()
	config := NewFunctionalTestConfig()
	consumer, err := NewConsumer(FunctionalTestEnv.KafkaBrokerAddrs, config)
	require.NoError(t, err)
	defer consumer.Close()

	pc, err := consumer.ConsumePartition(topic, partition, startOffset)
	require.NoError(t, err)
	defer pc.Close()

	result := make(map[string]string, count)
	ctx, cancel := context.WithTimeout(t.Context(), 15*time.Second)
	defer cancel()

	for len(result) < count {
		select {
		case msg := <-pc.Messages():
			require.NotNil(t, msg)
			result[string(msg.Key)] = string(msg.Value)
		case err := <-pc.Errors():
			require.NoError(t, err)
		case <-ctx.Done():
			return result
		}
	}
	return result
}

// TestJavaMurmur2PartitionerInterop verifies that Sarama's NewMurmur2Partitioner
// routes keyed messages to the same partitions as the Apache Kafka Java client's