Repository: apache/kafka
File: streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/DeadLetterQueueIntegrationTest.java (Java)

Code before:
        CLUSTER.stop();
    }

    private String applicationId;
    private static final int NUM_TOPIC_PARTITIONS = 3;
    private static final String INPUT_TOPIC = "inputTopic";
    private static final String OUTPUT_TOPIC = "outputTopic";
    private static final String DLQ_TOPIC = "dlqTopic";

    private static final AtomicInteger TEST_NUMBER = new AtomicInteger(0);

    private final List<KeyValue<String, String>> data = prepareData();

    @BeforeEach
    public void createTopics() throws Exception {
        applicationId = "appId-" + TEST_NUMBER.getAndIncrement();
        CLUSTER.deleteTopics(
            INPUT_TOPIC,
            OUTPUT_TOPIC,
            DLQ_TOPIC);
        CLUSTER.createTopic(INPUT_TOPIC, NUM_TOPIC_PARTITIONS, 1);
        CLUSTER.createTopic(OUTPUT_TOPIC, NUM_TOPIC_PARTITIONS, 1);
    }

    @Test
    public void shouldSendToDlqAnfFailFromDsl() throws Exception {

        try (final KafkaStreams streams = getDslStreams(LogAndFailProcessingExceptionHandler.class.getName())) {

            startApplicationAndWaitUntilRunning(streams);

            // Produce data to the input topic
            IntegrationTestUtils.produceKeyValuesSynchronously(
                INPUT_TOPIC,
                data,
                TestUtils.producerConfig(CLUSTER.bootstrapServers(), StringSerializer.class, StringSerializer.class),
                CLUSTER.time
            );

            // Consume the output records
            final List<ConsumerRecord<String, String>> outputRecords = readResult(OUTPUT_TOPIC, 1, StringDeserializer.class, StringDeserializer.class, 30000L);

            // Only the first record is available
            assertEquals(1, outputRecords.size(), "Only one record should be available in the output topic");
            assertEquals("value-1", outputRecords.get(0).value(), "Output record should be the first one");

            // Consume the DLQ records
            final List<ConsumerRecord<byte[], byte[]>> dlqRecords = readResult(DLQ_TOPIC, 1, ByteArrayDeserializer.class, ByteArrayDeserializer.class, 30000L);

            // Stream should be in ERROR state
            waitForApplicationState(singletonList(streams), KafkaStreams.State.ERROR, Duration.ofSeconds(30));

Review comment (bug):
Method name contains a typo: 'Anf' should be 'And'
```suggestion
    public void shouldSendToDlqAndFailFromDsl() throws Exception {
```

Code after:
    private String applicationId;
    private static final int NUM_TOPIC_PARTITIONS = 3;
    private static final String INPUT_TOPIC = "inputTopic";
    private static final String OUTPUT_TOPIC = "outputTopic";
    private static final String DLQ_TOPIC = "dlqTopic";

    private final List<KeyValue<String, String>> data = prepareData();
    private final List<KeyValue<String, byte[]>> bytesData = prepareBytesData();

    @BeforeEach
    public void createTopics(final TestInfo testInfo) throws Exception {
        applicationId = "appId-" + safeUniqueTestName(testInfo);
        cluster.deleteTopics(
            INPUT_TOPIC,
            OUTPUT_TOPIC,
            DLQ_TOPIC);
        cluster.createTopic(INPUT_TOPIC, NUM_TOPIC_PARTITIONS, 1);
        cluster.createTopic(OUTPUT_TOPIC, NUM_TOPIC_PARTITIONS, 1);
        cluster.createTopic(DLQ_TOPIC, NUM_TOPIC_PARTITIONS, 1);
    }

    @Test
    public void shouldSendToDlqAndFailFromDsl() throws Exception {
        try (final KafkaStreams streams = getDslStreams(LogAndFailProcessingExceptionHandler.class.getName())) {

            startApplicationAndWaitUntilRunning(streams);

            // Produce data to the input topic
            IntegrationTestUtils.produceKeyValuesSynchronously(
                INPUT_TOPIC,
                data,
                TestUtils.producerConfig(cluster.bootstrapServers(), StringSerializer.class, StringSerializer.class),
                cluster.time
            );

            // Consume the output records
            final List<ConsumerRecord<String, String>> outputRecords = readResult(OUTPUT_TOPIC, 1, StringDeserializer.class, StringDeserializer.class, 30000L);

            // Only the first record is available
            assertEquals(1, outputRecords.size(), "Only one record should be available in the output topic");
            assertEquals("value-1", outputRecords.get(0).value(), "Output record should be the first one");

            // Consume the DLQ records
            final List<ConsumerRecord<byte[], byte[]>> dlqRecords = readResult(DLQ_TOPIC, 1, ByteArrayDeserializer.class, ByteArrayDeserializer.class, 30000L);

            // Stream should be in ERROR state
            waitForApplicationState(singletonList(streams), KafkaStreams.State.ERROR, Duration.ofSeconds(30));

            assertEquals("key", new String(dlqRecords.get(0).key()), "Output record should be sent to DLQ topic");
            assertEquals("KABOOM", new String(dlqRecords.get(0).value()), "Output record should be sent to DLQ topic");