Function consumeWithSarama

functional_java_interop_test.go:85–112

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

}

func consumeWithSarama(t *testing.T, topic string, startOffset int64, count int) []string {
	t.Helper()
	config := NewFunctionalTestConfig()
	consumer, err := NewConsumer(FunctionalTestEnv.KafkaBrokerAddrs, config)
	require.NoError(t, err)
	defer consumer.Close()

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

	var messages []string
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	for i := 0; i < count; i++ {
		select {
		case msg := <-partitionConsumer.Messages():
			require.NotNil(t, msg)
			messages = append(messages, string(msg.Value))
		case err := <-partitionConsumer.Errors():
			require.NoError(t, err)
		case <-ctx.Done():
			require.Fail(t, "timeout waiting for messages")
		}
	}
	return messages
}

func produceWithSarama(t *testing.T, topic string, codec CompressionCodec, messages []string) {
	t.Helper()