@Test
    public void testFilterOnTombstonesWithSinkConnector() throws Exception {
        Map<String, Long> observedRecords = observeRecords();

        // create test topics
        String topic = "foo-topic";
        int numRecords = NUM_RECORDS_PRODUCED;
        connect.kafka().createTopic(topic, NUM_TOPIC_PARTITIONS);

        // setup up props for the sink connector
        Map<String, String> props = new HashMap<>();
        props.put("name", CONNECTOR_NAME);
        props.put(CONNECTOR_CLASS_CONFIG, SINK_CONNECTOR_CLASS_NAME);
        props.put(TASKS_MAX_CONFIG, String.valueOf(NUM_TASKS));
        props.put(TOPICS_CONFIG, String.join(",", topic));
        props.put(KEY_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
        props.put(VALUE_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
        props.put(TRANSFORMS_CONFIG, "filter");
        props.put(TRANSFORMS_CONFIG + ".filter.type", Filter.class.getSimpleName());
        props.put(TRANSFORMS_CONFIG + ".filter.predicate", "barPredicate");
        props.put(PREDICATES_CONFIG, "barPredicate");
        props.put(PREDICATES_CONFIG + ".barPredicate.type", RecordIsTombstone.class.getSimpleName());

        // expect only half the records to be consumed by the connector
        connectorHandle.expectedCommits(numRecords / 2);
        connectorHandle.expectedRecords(numRecords / 2);

        // start a sink connector
        connect.configureConnector(CONNECTOR_NAME, props);
        assertConnectorRunning();

        // produce some messages into source topic partitions
        for (int i = 0; i < numRecords; i++) {
            connect.kafka().produce(topic, i % NUM_TOPIC_PARTITIONS, "key", i % 2 == 0 ? "simple-message-value-" + i : null);
        }

        // consume all records from the source topic or fail, to ensure that they were correctly produced.
        assertEquals(
                numRecords,
                connect.kafka().consume(numRecords, RECORD_TRANSFER_DURATION_MS, topic).count(),
                "Unexpected number of records consumed"
        );

        // wait for the connector tasks to consume all records.
        connectorHandle.awaitRecords(RECORD_TRANSFER_DURATION_MS);

        // wait for the connector tasks to commit all records.
        connectorHandle.awaitCommits(RECORD_TRANSFER_DURATION_MS);

        Map<String, Long> expectedRecordCounts = singletonMap(topic, (long) (numRecords / 2));
        assertObservedRecords(observedRecords, expectedRecordCounts);

        // delete connector
        connect.deleteConnector(CONNECTOR_NAME);
    }
Test the {@link Filter} transformer with a
{@link RecordIsTombstone} predicate on a sink connector.
connect/runtime/src/test/java/org/apache/kafka/connect/integration/TransformationIntegrationTest.java