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

Code before:
            // Only the first and third records are available
            assertEquals(2, 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");
            assertEquals("value-2", outputRecords.get(1).value(), "Output record should be the third one");

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

            // Stream should be in RUNNING state
            assertThrows(AssertionError.class, () -> waitForApplicationState(singletonList(streams), KafkaStreams.State.ERROR, Duration.ofSeconds(10)));
            waitForApplicationState(singletonList(streams), KafkaStreams.State.RUNNING, Duration.ofSeconds(5));

            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");

            assertEquals("java.lang.RuntimeException: KABOOM", new String(dlqRecords.get(0).headers().lastHeader(HEADER_ERRORS_EXCEPTION_NAME).value()));
            assertEquals("KABOOM", new String(dlqRecords.get(0).headers().lastHeader(HEADER_ERRORS_EXCEPTION_MESSAGE_NAME).value()));
            assertTrue(new String(dlqRecords.get(0).headers().lastHeader(HEADER_ERRORS_STACKTRACE_NAME).value()).contains("org.apache.kafka.streams.kstream.internals.KStreamMapValues$KStreamMapProcessor.process"));
            assertEquals(INPUT_TOPIC, new String(dlqRecords.get(0).headers().lastHeader(HEADER_ERRORS_TOPIC_NAME).value()));
            assertEquals("1", new String(dlqRecords.get(0).headers().lastHeader(HEADER_ERRORS_PARTITION_NAME).value()));
            assertEquals("1", new String(dlqRecords.get(0).headers().lastHeader(HEADER_ERRORS_OFFSET_NAME).value()));
        }
    }

    @Test
    public void shouldSendToDlqAnfFailFromProcessorAPI() throws Exception {

        try (final KafkaStreams streams = getProcessorAPIStreams(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 shouldSendToDlqAndFailFromProcessorAPI() throws Exception {
```

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

            // Stream should be in RUNNING state
            assertThrows(AssertionError.class, () -> waitForApplicationState(singletonList(streams), KafkaStreams.State.ERROR, Duration.ofSeconds(10)));
            waitForApplicationState(singletonList(streams), KafkaStreams.State.RUNNING, Duration.ofSeconds(5));

            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");

            assertEquals("java.lang.RuntimeException: KABOOM", new String(dlqRecords.get(0).headers().lastHeader(HEADER_ERRORS_EXCEPTION_NAME).value()));
            assertEquals("KABOOM", new String(dlqRecords.get(0).headers().lastHeader(HEADER_ERRORS_EXCEPTION_MESSAGE_NAME).value()));
            assertTrue(new String(dlqRecords.get(0).headers().lastHeader(HEADER_ERRORS_STACKTRACE_NAME).value()).contains("org.apache.kafka.streams.kstream.internals.KStreamMapValues$KStreamMapProcessor.process"));
            assertEquals(INPUT_TOPIC, new String(dlqRecords.get(0).headers().lastHeader(HEADER_ERRORS_TOPIC_NAME).value()));
            assertEquals("1", new String(dlqRecords.get(0).headers().lastHeader(HEADER_ERRORS_PARTITION_NAME).value()));
            assertEquals("1", new String(dlqRecords.get(0).headers().lastHeader(HEADER_ERRORS_OFFSET_NAME).value()));
        }
    }

    @Test
    public void shouldSendToDlqAndFailFromProcessorAPI() throws Exception {
        try (final KafkaStreams streams = getProcessorAPIStreams(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");

            assertEquals("java.lang.RuntimeException: KABOOM", new String(dlqRecords.get(0).headers().lastHeader(HEADER_ERRORS_EXCEPTION_NAME).value()));
            assertEquals("KABOOM", new String(dlqRecords.get(0).headers().lastHeader(HEADER_ERRORS_EXCEPTION_MESSAGE_NAME).value()));