Function produceWithJava

functional_java_interop_test.go:35–83

(t *testing.T, topic string, codec CompressionCodec, messages []string)

}

func produceWithJava(t *testing.T, topic string, codec CompressionCodec, messages []string) {
	t.Helper()
	producerPath := fmt.Sprintf("/opt/kafka-%s/bin/kafka-console-producer.sh", FunctionalTestEnv.KafkaVersion)
	args := append(
		[]string{"compose", "exec", "-T", brokerContainer, producerPath},
		javaProducerArgs(topic, codec)...,
	)
	cmd := exec.Command("docker", args...)

	stdin, err := cmd.StdinPipe()
	require.NoError(t, err)

	stderr, err := cmd.StderrPipe()
	require.NoError(t, err)

	var stderrOutput strings.Builder
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		s := bufio.NewScanner(stderr)
		for s.Scan() {
			stderrOutput.WriteString(s.Text() + "\n")
		}
	}()

	require.NoError(t, cmd.Start())

	for _, msg := range messages {
		_, err := fmt.Fprintln(stdin, msg)
		if err != nil {
			stdin.Close()
			waitErr := cmd.Wait()
			wg.Wait()
			if waitErr != nil {
				err = fmt.Errorf("failed to write message: %w; Java producer failed: %w; stderr: %s", err, waitErr, stderrOutput.String())
			}
		}
		require.NoError(t, err)
	}
	stdin.Close()

	err = cmd.Wait()
	wg.Wait()
	if err != nil {
		t.Logf("Java producer stderr: %s", stderrOutput.String())
		require.NoError(t, err, "Java producer failed")
	}
}

func consumeWithSarama(t *testing.T, topic string, startOffset int64, count int) []string {
	t.Helper()