Repository: apache/kafka
File: clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ConsumerWithLegacyMessageFormatIntegrationTest.java (Java)

Code before:
                magicValue,
                Compression.of(CompressionType.NONE).build(),
                TimestampType.CREATE_TIME,
                0L,
                RecordBatch.NO_TIMESTAMP,
                RecordBatch.NO_PRODUCER_ID,
                RecordBatch.NO_PRODUCER_EPOCH,
                0,
                false,
                RecordBatch.NO_PARTITION_LEADER_EPOCH
        );

        records.forEach(builder::append);

        cluster.brokers().values().stream()
                .filter(b -> b.config().brokerId() == brokerId)
                .forEach(b -> {
                    UnifiedLog unifiedLog = b.replicaManager().logManager().getLog(tp, false).get();
                    unifiedLog.appendAsLeaderWithRecordVersion(builder.build(), 0, RecordVersion.lookup(magicValue));
                    try {
                        // Default isolation.level is read_uncommitted. It makes Partition#fetchOffsetForTimestamp to return UnifiedLog#highWatermark,
                        // so increasing high watermark to make it return the correct offset.
                        unifiedLog.maybeIncrementHighWatermark(unifiedLog.logEndOffsetMetadata());
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                });
    }

    private void createTopicWithAssignment(String topic, Map<Integer, List<Integer>> assignment) throws InterruptedException {
        try (Admin admin = cluster.admin()) {
            NewTopic newTopic = new NewTopic(topic, assignment);
            admin.createTopics(Collections.singletonList(newTopic));
            cluster.waitForTopic(topic, assignment.size());
        }
    }

    private void sendRecords(int numRecords, TopicPartition topicPartition, long startingTimestamp) {
        try (Producer<byte[], byte[]> producer = cluster.producer()) {
            for (int i = 0; i < numRecords; i++) {
                long timestamp = startingTimestamp + i;
                byte[] key = ("key " + i).getBytes();
                byte[] value = ("value " + i).getBytes();

                ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(
                        topicPartition.topic(),
                        topicPartition.partition(),
                        timestamp,
                        key,
                        value
                );

Review comment (suggestion):
```suggestion
                    // Default isolation.level is read_uncommitted. It makes Partition#fetchOffsetForTimestamp to return UnifiedLog#highWatermark,
                    // so increasing high watermark to make it return the correct offset.
                    assertDoesNotThrow(() -> unifiedLog.maybeIncrementHighWatermark(unifiedLog.logEndOffsetMetadata()));
```

Code after:
                Compression.of(CompressionType.NONE).build(),
                TimestampType.CREATE_TIME,
                0L,
                RecordBatch.NO_TIMESTAMP,
                RecordBatch.NO_PRODUCER_ID,
                RecordBatch.NO_PRODUCER_EPOCH,
                0,
                false,
                RecordBatch.NO_PARTITION_LEADER_EPOCH
        );

        records.forEach(builder::append);

        cluster.brokers().values().stream()
                .filter(b -> b.config().brokerId() == brokerId)
                .forEach(b -> {
                    UnifiedLog unifiedLog = b.replicaManager().logManager().getLog(tp, false).get();
                    unifiedLog.appendAsLeaderWithRecordVersion(builder.build(), 0, RecordVersion.lookup(magicValue));
                    // Default isolation.level is read_uncommitted. It makes Partition#fetchOffsetForTimestamp to return UnifiedLog#highWatermark,
                    // so increasing high watermark to make it return the correct offset.
                    assertDoesNotThrow(() -> unifiedLog.maybeIncrementHighWatermark(unifiedLog.logEndOffsetMetadata()));
                });
    }

    private void createTopicWithAssignment(String topic, Map<Integer, List<Integer>> assignment) throws InterruptedException {
        try (Admin admin = cluster.admin()) {
            NewTopic newTopic = new NewTopic(topic, assignment);
            admin.createTopics(List.of(newTopic));
            cluster.waitForTopic(topic, assignment.size());
        }
    }

    @BeforeEach
    public void setupTopics() throws InterruptedException {
        cluster.createTopic(topic1, 2, (short) 1);
        createTopicWithAssignment(topic2, Map.of(0, List.of(0), 1, List.of(1)));
        createTopicWithAssignment(topic3, Map.of(0, List.of(0), 1, List.of(1)));

        // v2 message format for topic1
        ClientsTestUtils.sendRecords(cluster, t1p0, 100, 0);
        ClientsTestUtils.sendRecords(cluster, t1p1, 100, 0);
        // v0 message format for topic2
        appendLegacyRecords(100, t2p0, 0, RecordBatch.MAGIC_VALUE_V0);
        appendLegacyRecords(100, t2p1, 1, RecordBatch.MAGIC_VALUE_V0);
        // v1 message format for topic3
        appendLegacyRecords(100, t3p0, 0, RecordBatch.MAGIC_VALUE_V1);
        appendLegacyRecords(100, t3p1, 1, RecordBatch.MAGIC_VALUE_V1);
    }

    @ClusterTest
    public void testOffsetsForTimesWithClassicConsumer() {