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

Code before:

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

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

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

        for (int i = 0; i < NUM_RECORDS; 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 (refactor):
The variable `NUM_RECORDS` alaways 1, we should refactor these methods.

Code after:
        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.");

        ConsumerRecord<String, String> record = records.get(0);
        assertEquals(TOPIC, record.topic());
        assertEquals(PARTITION, record.partition());
        assertEquals(0, record.offset());
    }
}