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

Code before:
        // Cluster ID in deserializer before deserialize
        Assertions.assertNotEquals(MockDeserializer.clusterIdBeforeDeserialize.get(), MockDeserializer.noClusterId);
        Assertions.assertNotNull(MockDeserializer.clusterMeta);
        isValidClusterId(MockDeserializer.clusterMeta.get().clusterId());
        Assertions.assertEquals(
            MockDeserializer.clusterIdBeforeDeserialize.get().clusterId(),
            MockDeserializer.clusterMeta.get().clusterId()
        );

        Assertions.assertNotNull(MockConsumerMetricsReporter.CLUSTER_META);
        isValidClusterId(MockConsumerMetricsReporter.CLUSTER_META.get().clusterId());

        // All components received the same cluster id
        String id = MockProducerInterceptor.CLUSTER_META.get().clusterId();
        Assertions.assertEquals(id, MockSerializer.CLUSTER_META.get().clusterId());
        Assertions.assertEquals(id, MockProducerMetricsReporter.CLUSTER_META.get().clusterId());
        Assertions.assertEquals(id, MockConsumerInterceptor.CLUSTER_META.get().clusterId());
        Assertions.assertEquals(id, MockDeserializer.clusterMeta.get().clusterId());
        Assertions.assertEquals(id, MockConsumerMetricsReporter.CLUSTER_META.get().clusterId());
        Assertions.assertEquals(id, MockBrokerMetricsReporter.CLUSTER_META.get().clusterId());

        MockConsumerInterceptor.resetCounters();
        MockProducerInterceptor.resetCounters();
    }

    private static void sendRecords(Producer<byte[], byte[]> producer, int numRecords, TopicPartition tp) throws Exception {
        for (int i = 0; i < numRecords; i++) {
            ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(tp.topic(), tp.partition(), ("" + i).getBytes(), ("" + i).getBytes());
            producer.send(record).get();
        }
    }

    private void consumeRecords(Consumer<byte[], byte[]> consumer, int numRecords) throws InterruptedException {
        List<ConsumerRecord<byte[], byte[]>> records = new ArrayList<>();
        TestUtils.waitForCondition(() -> {
            consumer.poll(Duration.ofMillis(100)).forEach(records::add);
            return records.size() >= numRecords;
        }, 60000, "Timed out before consuming expected " + numRecords + " records.");

        for (int i = 0; i < numRecords; i++) {
            ConsumerRecord<byte[], byte[]> record = records.get(i);
            Assertions.assertEquals(TOPIC, record.topic());
            Assertions.assertEquals(PARTITION, record.partition());
            Assertions.assertEquals(i, record.offset());
        }
    }
}

Review comment (suggestion):
we don't need `int numRecords, TopicPartition tp`, right? this method can use constants directly

Code after:
            consumer.seek(TP, 0);
            // consume and verify that values are modified by interceptors
            consumeRecord(consumer);
        }

        // Check that cluster id is present after the first poll call.
        assertNotEquals(MockConsumerInterceptor.CLUSTER_ID_BEFORE_ON_CONSUME.get(), MockConsumerInterceptor.NO_CLUSTER_ID);
        assertNotNull(MockConsumerInterceptor.CLUSTER_META.get());
        isValidClusterId(MockConsumerInterceptor.CLUSTER_META.get().clusterId());
        assertEquals(
            MockConsumerInterceptor.CLUSTER_ID_BEFORE_ON_CONSUME.get().clusterId(),
            MockConsumerInterceptor.CLUSTER_META.get().clusterId()
        );

        assertNotEquals(MockDeserializer.clusterIdBeforeDeserialize.get(), MockDeserializer.noClusterId);
        assertNotNull(MockDeserializer.clusterMeta);
        isValidClusterId(MockDeserializer.clusterMeta.get().clusterId());
        assertEquals(
            MockDeserializer.clusterIdBeforeDeserialize.get().clusterId(),
            MockDeserializer.clusterMeta.get().clusterId()
        );

        ClusterResource consumerClusterResource = MockCommonMetricsReporter.CLUSTER_RESOURCE_MAP.get(CONSUMER_CLIENT_ID);
        assertNotNull(consumerClusterResource);
        isValidClusterId(consumerClusterResource.clusterId());

        // Make sure everyone receives the same cluster id.
        String id = MockProducerInterceptor.CLUSTER_META.get().clusterId();
        assertEquals(id, MockSerializer.CLUSTER_META.get().clusterId());
        assertEquals(id, MockCommonMetricsReporter.CLUSTER_RESOURCE_MAP.get(PRODUCER_CLIENT_ID).clusterId());
        assertEquals(id, MockConsumerInterceptor.CLUSTER_META.get().clusterId());
        assertEquals(id, MockDeserializer.clusterMeta.get().clusterId());
        assertEquals(id, MockCommonMetricsReporter.CLUSTER_RESOURCE_MAP.get(CONSUMER_CLIENT_ID).clusterId());
        assertEquals(id, MockCommonMetricsReporter.CLUSTER_RESOURCE_MAP.get(clusterBrokerId).clusterId());
        assertEquals(id, MockCommonMetricsReporter.CLUSTER_RESOURCE_MAP.get(controllerId).clusterId());

        MockConsumerInterceptor.resetCounters();
        MockProducerInterceptor.resetCounters();
    }

    private static void sendRecord(Producer<String, String> producer) throws Exception {
        ProducerRecord<String, String> record = new ProducerRecord<>(TP.topic(), TP.partition(), "0", "0");
        producer.send(record).get();
    }

    private void consumeRecord(Consumer<String, String> consumer) throws InterruptedException {
        List<ConsumerRecord<String, String>> records = new ArrayList<>();
        TestUtils.waitForCondition(() -> {
            consumer.poll(Duration.ofMillis(100)).forEach(records::add);
            return !records.isEmpty();
        }, 60000, "Timed out before consuming expected record.");