Repository: apache/kafka
File: streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/IQv2VersionedStoreIntegrationTest.java (Java)

Code before:
        // Delete and recreate the topic to ensure clean state for each test
        CLUSTER.deleteTopic(INPUT_TOPIC_NAME);
        CLUSTER.createTopic(INPUT_TOPIC_NAME, 1, 1);
        
        // Set up fresh test data
        final Properties producerProps = new Properties();
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        try (final KafkaProducer<Integer, Integer> producer = new KafkaProducer<>(producerProps)) {
            producer.send(new ProducerRecord<>(INPUT_TOPIC_NAME, 0,  RECORD_TIMESTAMPS[0], RECORD_KEY, RECORD_VALUES[0])).get();
            producer.send(new ProducerRecord<>(INPUT_TOPIC_NAME, 0,  RECORD_TIMESTAMPS[1], RECORD_KEY, RECORD_VALUES[1])).get();
            producer.send(new ProducerRecord<>(INPUT_TOPIC_NAME, 0,  RECORD_TIMESTAMPS[2], RECORD_KEY, RECORD_VALUES[2])).get();
            producer.send(new ProducerRecord<>(INPUT_TOPIC_NAME, 0,  RECORD_TIMESTAMPS[3], RECORD_KEY, RECORD_VALUES[3])).get();
        }
        inputPosition = Position.emptyPosition();
        inputPosition = inputPosition.withComponent(INPUT_TOPIC_NAME, 0, 3);
    }

    private void setup(final String groupProtocol) {
        this.groupProtocol = groupProtocol;
        final StreamsBuilder builder = new StreamsBuilder();
        builder.table(INPUT_TOPIC_NAME,
            Materialized.as(Stores.persistentVersionedKeyValueStore(STORE_NAME, HISTORY_RETENTION, SEGMENT_INTERVAL)));
        final Properties configs = new Properties();
        configs.put(StreamsConfig.APPLICATION_ID_CONFIG, "app-" + System.nanoTime() + "-" + groupProtocol);
        configs.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
        configs.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.IntegerSerde.class.getName());
        configs.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.IntegerSerde.class.getName());
        configs.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, groupProtocol);
        kafkaStreams = IntegrationTestUtils.getStartedStreams(configs, builder, true);
    }

    @AfterEach
    public void afterTest() {
        if (kafkaStreams != null) {
            kafkaStreams.close(Duration.ofSeconds(60));
            kafkaStreams.cleanUp();
        }
    }

    @AfterAll
    public static void after() {
        CLUSTER.stop();
    }

    private static Stream<Arguments> groupProtocolParameters() {
        return Stream.of(
            Arguments.of("classic", "CLASSIC protocol"),
            Arguments.of("streams", "STREAMS protocol")
        );

Review comment (suggestion):
Using System.nanoTime() for APPLICATION_ID_CONFIG is not recommended as it can result in non-deterministic behavior and potential collisions. Consider using a more deterministic approach like combining test name with group protocol.
```suggestion
        configs.put(StreamsConfig.APPLICATION_ID_CONFIG, getClass().getSimpleName() + "-" + groupProtocol);
```

Code after:
    @BeforeEach
    public void beforeEach() throws Exception {
        // Delete and recreate the topic to ensure clean state for each test
        CLUSTER.deleteTopic(INPUT_TOPIC_NAME);
        CLUSTER.createTopic(INPUT_TOPIC_NAME, 1, 1);
        
        // Set up fresh test data
        final Properties producerProps = new Properties();
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        try (final KafkaProducer<Integer, Integer> producer = new KafkaProducer<>(producerProps)) {
            producer.send(new ProducerRecord<>(INPUT_TOPIC_NAME, 0,  RECORD_TIMESTAMPS[0], RECORD_KEY, RECORD_VALUES[0])).get();
            producer.send(new ProducerRecord<>(INPUT_TOPIC_NAME, 0,  RECORD_TIMESTAMPS[1], RECORD_KEY, RECORD_VALUES[1])).get();
            producer.send(new ProducerRecord<>(INPUT_TOPIC_NAME, 0,  RECORD_TIMESTAMPS[2], RECORD_KEY, RECORD_VALUES[2])).get();
            producer.send(new ProducerRecord<>(INPUT_TOPIC_NAME, 0,  RECORD_TIMESTAMPS[3], RECORD_KEY, RECORD_VALUES[3])).get();
        }
        inputPosition = Position.emptyPosition().withComponent(INPUT_TOPIC_NAME, 0, 3);
    }

    private void setup(final String groupProtocol, final TestInfo testInfo) {
        this.groupProtocol = groupProtocol;
        final StreamsBuilder builder = new StreamsBuilder();
        builder.table(INPUT_TOPIC_NAME,
            Materialized.as(Stores.persistentVersionedKeyValueStore(STORE_NAME, HISTORY_RETENTION, SEGMENT_INTERVAL)));
        final Properties configs = new Properties();
        final String safeTestName = safeUniqueTestName(testInfo);
        configs.put(StreamsConfig.APPLICATION_ID_CONFIG, "app-" + safeTestName);
        configs.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
        configs.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.IntegerSerde.class.getName());
        configs.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.IntegerSerde.class.getName());
        configs.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, groupProtocol);
        kafkaStreams = IntegrationTestUtils.getStartedStreams(configs, builder, true);
    }

    @AfterEach
    public void afterTest() {
        if (kafkaStreams != null) {
            kafkaStreams.close(Duration.ofSeconds(60));
            kafkaStreams.cleanUp();
        }
    }

    @AfterAll
    public static void after() {
        CLUSTER.stop();
    }

    private static Stream<Arguments> groupProtocolParameters() {
        return Stream.of(
            Arguments.of("classic", "CLASSIC protocol"),