private Map<String, Long> observeRecords() {
        Map<String, Long> observedRecords = new HashMap<>();
        // record all the record we see
        connectorHandle.taskHandle(CONNECTOR_NAME + "-0",
            record -> observedRecords.compute(record.topic(),
                (key, value) -> value == null ? 1 : value + 1));
        return observedRecords;
    }
Test the {@link Filter} transformer with a
{@link TopicNameMatches} predicate on a sink connector.
connect/runtime/src/test/java/org/apache/kafka/connect/integration/TransformationIntegrationTest.java