@Test
    public void testFilterOnHasHeaderKeyWithSourceConnectorAndTopicCreation() throws Exception {
        // setup up props for the sink connector
        Map<String, String> props = new HashMap<>();
        props.put("name", CONNECTOR_NAME);
        props.put(CONNECTOR_CLASS_CONFIG, SOURCE_CONNECTOR_CLASS_NAME);
        props.put(TASKS_MAX_CONFIG, String.valueOf(NUM_TASKS));
        props.put("topic", "test-topic");
        props.put("throughput", String.valueOf(500));
        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", "headerPredicate");
        props.put(TRANSFORMS_CONFIG + ".filter.negate", "true");
        props.put(PREDICATES_CONFIG, "headerPredicate");
        props.put(PREDICATES_CONFIG + ".headerPredicate.type", HasHeaderKey.class.getSimpleName());
        props.put(PREDICATES_CONFIG + ".headerPredicate.name", "header-8");
        // custom topic creation is used, so there's no need to proactively create the test topic
        props.put(DEFAULT_TOPIC_CREATION_PREFIX + REPLICATION_FACTOR_CONFIG, String.valueOf(-1));
        props.put(DEFAULT_TOPIC_CREATION_PREFIX + PARTITIONS_CONFIG, String.valueOf(NUM_TOPIC_PARTITIONS));

        // expect all records to be produced by the connector
        connectorHandle.expectedRecords(NUM_RECORDS_PRODUCED);

        // expect all records to be produced by the connector
        connectorHandle.expectedCommits(NUM_RECORDS_PRODUCED);

        // validate the intended connector configuration, a valid config
        connect.assertions().assertExactlyNumErrorsOnConnectorConfigValidation(SOURCE_CONNECTOR_CLASS_NAME, props, 0,
                "Validating connector configuration produced an unexpected number or errors.");

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

        // wait for the connector tasks to produce enough records
        connectorHandle.awaitRecords(RECORD_TRANSFER_DURATION_MS);

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

        // consume all records from the source topic or fail, to ensure that they were correctly produced
        for (ConsumerRecord<byte[], byte[]> record : connect.kafka().consume(1, RECORD_TRANSFER_DURATION_MS, "test-topic")) {
            assertNotNull(record.headers().lastHeader("header-8"), "Expected header to exist");
        }

        // delete connector
        connect.deleteConnector(CONNECTOR_NAME);
    }
Test the {@link Filter} transformer with a {@link HasHeaderKey} predicate on a source connector.
Note that this test uses topic creation configs to allow the source connector to create
the topic when it tries to produce the first source record, instead of requiring the topic
to exist before the connector starts.
connect/runtime/src/test/java/org/apache/kafka/connect/integration/TransformationIntegrationTest.java