@Test
    public void testErrantRecordReporter() throws Exception {
        // create test topic
        connect.kafka().createTopic("test-topic");

        // setup connector config
        Map<String, String> props = new HashMap<>();
        props.put(CONNECTOR_CLASS_CONFIG, ErrantRecordSinkConnector.class.getSimpleName());
        props.put(TASKS_MAX_CONFIG, String.valueOf(NUM_TASKS));
        props.put(TOPICS_CONFIG, "test-topic");
        // Restrict the size of each poll so that the records are delivered across multiple polls
        props.put(CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX + ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "5");
        props.put(KEY_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
        props.put(VALUE_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());

        // log all errors, along with message metadata
        props.put(ERRORS_LOG_ENABLE_CONFIG, "true");
        props.put(ERRORS_LOG_INCLUDE_MESSAGES_CONFIG, "true");

        // produce bad messages into dead letter queue
        props.put(DLQ_TOPIC_NAME_CONFIG, DLQ_TOPIC);
        props.put(DLQ_CONTEXT_HEADERS_ENABLE_CONFIG, "true");
        props.put(DLQ_TOPIC_REPLICATION_FACTOR_CONFIG, "1");

        // tolerate all errors
        props.put(ERRORS_TOLERANCE_CONFIG, "all");

        // retry for up to one second
        props.put(ERRORS_RETRY_TIMEOUT_CONFIG, "1000");

        // set expected records to successfully reach the task
        connectorHandle.taskHandle(TASK_ID).expectedRecords(NUM_RECORDS_PRODUCED);

        connect.configureConnector(CONNECTOR_NAME, props);
        connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(CONNECTOR_NAME, NUM_TASKS,
            "Connector tasks did not start in time.");

        waitForCondition(this::checkForPartitionAssignment,
            CONNECTOR_SETUP_DURATION_MS,
            "Connector task was not assigned a partition.");

        // produce some strings into test topic
        for (int i = 0; i < NUM_RECORDS_PRODUCED; i++) {
            connect.kafka().produce("test-topic", "key-" + i, "value-" + i);
        }

        // consume all records from test topic
        log.info("Consuming records from test topic");
        int i = 0;
        for (ConsumerRecord<byte[], byte[]> rec : connect.kafka().consume(NUM_RECORDS_PRODUCED, CONSUME_MAX_DURATION_MS, "test-topic")) {
            String k = new String(rec.key());
            String v = new String(rec.value());
            log.debug("Consumed record (key='{}', value='{}') from topic {}", k, v, rec.topic());
            assertEquals(k, "key-" + i, "Unexpected key");
            assertEquals(v, "value-" + i, "Unexpected value");
            i++;
        }

        // wait for records to reach the task
        connectorHandle.taskHandle(TASK_ID).awaitRecords(CONSUME_MAX_DURATION_MS);

        // consume failed records from dead letter queue topic
        log.info("Consuming records from test topic");
        Set<String> keys = new HashSet<>();
        for (ConsumerRecord<byte[], byte[]> rec : connect.kafka().consume(NUM_RECORDS_PRODUCED, CONSUME_MAX_DURATION_MS, DLQ_TOPIC)) {
            String k = new String(rec.key());
            keys.add(k);
        }
        assertEquals(NUM_RECORDS_PRODUCED, keys.size());

        connect.deleteConnector(CONNECTOR_NAME);
        connect.assertions().assertConnectorDoesNotExist(CONNECTOR_NAME,
            "Connector wasn't deleted in time.");
    }
Integration test for the different error handling policies in Connect (namely, retry policies, skipping bad records,
and dead letter queues).
connect/runtime/src/test/java/org/apache/kafka/connect/integration/ErrorHandlingIntegrationTest.java