Repository: apache/kafka
File: server/src/test/java/org/apache/kafka/api/EndToEndClusterIdTest.java (Java)

Code before:
        testEndToEnd(GroupProtocol.CLASSIC);
    }

    @ClusterTest
    public void testEndToEndWithConsumerProtocol() throws Exception {
        testEndToEnd(GroupProtocol.CONSUMER);
    }

    public void testEndToEnd(GroupProtocol groupProtocol) throws Exception {
        MockConsumerInterceptor.resetCounters();
        MockProducerInterceptor.resetCounters();

        ClusterResource brokerClusterResource = MockCommonMetricsReporter.CLUSTER_RESOURCE_MAP.get(clusterBrokerId);
        assertNotNull(brokerClusterResource);
        isValidClusterId(brokerClusterResource.clusterId());
        ClusterResource controllerClusterResource = MockCommonMetricsReporter.CLUSTER_RESOURCE_MAP.get(controllerId);
        assertNotNull(controllerClusterResource);
        isValidClusterId(controllerClusterResource.clusterId());

        Map<String, Object> producerConfig = Map.of(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, MockProducerInterceptor.class.getName(),
            "mock.interceptor.append", "mock",
            ProducerConfig.METRIC_REPORTER_CLASSES_CONFIG, MockCommonMetricsReporter.class.getName(),
            ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, MockSerializer.class.getName(),
            ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, MockSerializer.class.getName(),
            ProducerConfig.CLIENT_ID_CONFIG, PRODUCER_CLIENT_ID);
        try (var producer = clusterInstance.<byte[], byte[]>producer(producerConfig)) {
            // Send one record and make sure clusterId is set after sending and before onAcknowledgement
            sendRecord(producer);
        }
        assertNotEquals(MockProducerInterceptor.CLUSTER_ID_BEFORE_ON_ACKNOWLEDGEMENT.get(), MockProducerInterceptor.NO_CLUSTER_ID);
        assertNotNull(MockProducerInterceptor.CLUSTER_META.get());
        assertEquals(
            MockProducerInterceptor.CLUSTER_ID_BEFORE_ON_ACKNOWLEDGEMENT.get().clusterId(),
            MockProducerInterceptor.CLUSTER_META.get().clusterId()
        );
        isValidClusterId(MockProducerInterceptor.CLUSTER_META.get().clusterId());

        // Make sure the serializer sees Cluster ID before serialize method
        assertNotEquals(MockSerializer.CLUSTER_ID_BEFORE_SERIALIZE.get(), MockSerializer.NO_CLUSTER_ID);
        assertNotNull(MockSerializer.CLUSTER_META.get());
        isValidClusterId(MockSerializer.CLUSTER_META.get().clusterId());

        ClusterResource producerClusterResource = MockCommonMetricsReporter.CLUSTER_RESOURCE_MAP.get(PRODUCER_CLIENT_ID);
        assertNotNull(producerClusterResource);
        isValidClusterId(producerClusterResource.clusterId());

        Map<String, Object> consumerConfig = Map.of(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, MockConsumerInterceptor.class.getName(),
            ConsumerConfig.METRIC_REPORTER_CLASSES_CONFIG, MockCommonMetricsReporter.class.getName(),
            ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, MockDeserializer.class.getName(),
            ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, MockDeserializer.class.getName(),
            ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol.name(),

Review comment (suggestion):
Please use `String` instead as `MockProducerInterceptor` is using `String`

Code after:
    public void testEndToEndWithClassicProtocol() throws Exception {
        testEndToEnd(GroupProtocol.CLASSIC);
    }

    @ClusterTest
    public void testEndToEndWithConsumerProtocol() throws Exception {
        testEndToEnd(GroupProtocol.CONSUMER);
    }

    public void testEndToEnd(GroupProtocol groupProtocol) throws Exception {
        MockConsumerInterceptor.resetCounters();
        MockProducerInterceptor.resetCounters();

        ClusterResource brokerClusterResource = MockCommonMetricsReporter.CLUSTER_RESOURCE_MAP.get(clusterBrokerId);
        assertNotNull(brokerClusterResource);
        isValidClusterId(brokerClusterResource.clusterId());
        ClusterResource controllerClusterResource = MockCommonMetricsReporter.CLUSTER_RESOURCE_MAP.get(controllerId);
        assertNotNull(controllerClusterResource);
        isValidClusterId(controllerClusterResource.clusterId());

        Map<String, Object> producerConfig = Map.of(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, MockProducerInterceptor.class.getName(),
            "mock.interceptor.append", "mock",
            ProducerConfig.METRIC_REPORTER_CLASSES_CONFIG, MockCommonMetricsReporter.class.getName(),
            ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, MockSerializer.class.getName(),
            ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, MockSerializer.class.getName(),
            ProducerConfig.CLIENT_ID_CONFIG, PRODUCER_CLIENT_ID);
        try (var producer = clusterInstance.<String, String>producer(producerConfig)) {
            // Send one record and make sure clusterId is set after sending and before onAcknowledgement
            sendRecord(producer);
        }
        assertNotEquals(MockProducerInterceptor.CLUSTER_ID_BEFORE_ON_ACKNOWLEDGEMENT.get(), MockProducerInterceptor.NO_CLUSTER_ID);
        assertNotNull(MockProducerInterceptor.CLUSTER_META.get());
        assertEquals(
            MockProducerInterceptor.CLUSTER_ID_BEFORE_ON_ACKNOWLEDGEMENT.get().clusterId(),
            MockProducerInterceptor.CLUSTER_META.get().clusterId()
        );
        isValidClusterId(MockProducerInterceptor.CLUSTER_META.get().clusterId());

        // Make sure the serializer sees Cluster ID before serialize method
        assertNotEquals(MockSerializer.CLUSTER_ID_BEFORE_SERIALIZE.get(), MockSerializer.NO_CLUSTER_ID);
        assertNotNull(MockSerializer.CLUSTER_META.get());
        isValidClusterId(MockSerializer.CLUSTER_META.get().clusterId());

        ClusterResource producerClusterResource = MockCommonMetricsReporter.CLUSTER_RESOURCE_MAP.get(PRODUCER_CLIENT_ID);
        assertNotNull(producerClusterResource);
        isValidClusterId(producerClusterResource.clusterId());

        Map<String, Object> consumerConfig = Map.of(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, MockConsumerInterceptor.class.getName(),
            ConsumerConfig.METRIC_REPORTER_CLASSES_CONFIG, MockCommonMetricsReporter.class.getName(),
            ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, MockDeserializer.class.getName(),
            ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, MockDeserializer.class.getName(),