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

        // create test topics
        String fooTopic = "foo-topic";
        String barTopic = "bar-topic";
        int numFooRecords = NUM_RECORDS_PRODUCED;
        int numBarRecords = NUM_RECORDS_PRODUCED;
        connect.kafka().createTopic(fooTopic, NUM_TOPIC_PARTITIONS);
        connect.kafka().createTopic(barTopic, 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(",", fooTopic, barTopic));
        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", TopicNameMatches.class.getSimpleName());
        props.put(PREDICATES_CONFIG + ".barPredicate.pattern", "bar-.*");

        // expect all records to be consumed by the connector
        connectorHandle.expectedRecords(numFooRecords);

        // expect all records to be consumed by the connector
        connectorHandle.expectedCommits(numFooRecords);

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

        // produce some messages into source topic partitions
        for (int i = 0; i < numBarRecords; i++) {
            connect.kafka().produce(barTopic, i % NUM_TOPIC_PARTITIONS, "key", "simple-message-value-" + i);
        }
        for (int i = 0; i < numFooRecords; i++) {
            connect.kafka().produce(fooTopic, i % NUM_TOPIC_PARTITIONS, "key", "simple-message-value-" + i);
        }

        // consume all records from the source topic or fail, to ensure that they were correctly produced.
        assertEquals(
                numFooRecords,
                connect.kafka().consume(numFooRecords, RECORD_TRANSFER_DURATION_MS, fooTopic).count(),
                "Unexpected number of records consumed"
        );
        assertEquals(
                numBarRecords,
                connect.kafka().consume(numBarRecords, RECORD_TRANSFER_DURATION_MS, barTopic).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);

        // Assert that we didn't see any baz
        Map<String, Long> expectedRecordCounts = singletonMap(fooTopic, (long) numFooRecords);
        assertObservedRecords(observedRecords, expectedRecordCounts);

        // delete connector
        connect.deleteConnector(CONNECTOR_NAME);
    }
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