Repository: apache/kafka
File: storage/src/test/java/org/apache/kafka/tiered/storage/integration/BaseDeleteSegmentsTest.java (Java)

Code before:

    @Override
    protected void writeTestSpecifications(TieredStorageTestBuilder builder) {
        final Integer broker0 = 0;
        final String topicA = "topicA";
        final Integer p0 = 0;
        final Integer partitionCount = 1;
        final Integer replicationFactor = 1;
        final Integer maxBatchCountPerSegment = 1;
        final Map<Integer, List<Integer>> replicaAssignment = null;
        final boolean enableRemoteLogStorage = true;
        final int beginEpoch = 0;
        final long startOffset = 3;

        // Create topicA with 1 partition, 1 RF and enabled with remote storage.
        builder.createTopic(topicA, partitionCount, replicationFactor, maxBatchCountPerSegment, replicaAssignment,
                        enableRemoteLogStorage)
                // produce events to partition 0 and expect 3 segments to be offloaded
                .expectSegmentToBeOffloaded(broker0, topicA, p0, 0, new KeyValueSpec("k0", "v0"))
                .expectSegmentToBeOffloaded(broker0, topicA, p0, 1, new KeyValueSpec("k1", "v1"))
                .expectSegmentToBeOffloaded(broker0, topicA, p0, 2, new KeyValueSpec("k2", "v2"))
                .expectEarliestLocalOffsetInLogDirectory(topicA, p0, 3L)
                .produceWithTimestamp(topicA, p0, new KeyValueSpec("k0", "v0"), new KeyValueSpec("k1", "v1"),
                        // Using a future timestamp to test the log segment rolling mechanism.
                        // This simulates the passage of time, triggering time-based log segment rolling conditions
                        // When remote storage is enabled, this helps test how segments are marked for deletion
                        new KeyValueSpec("k2", "v2"), new KeyValueSpec("k3", "v3", System.currentTimeMillis() + TimeUnit.HOURS.toMillis(1)))
                // update the topic config such that it triggers the deletion of segments
                .updateTopicConfig(topicA, configsToBeAdded(), Collections.emptyList())
                // expect that the three offloaded remote log segments are deleted
                .expectDeletionInRemoteStorage(broker0, topicA, p0, DELETE_SEGMENT, 3)
                .waitForRemoteLogSegmentDeletion(topicA)
                // expect that the leader epoch checkpoint is updated
                .expectLeaderEpochCheckpoint(broker0, topicA, p0, beginEpoch, startOffset)
                // consume from the beginning of the topic to read data from local and remote storage
                .expectFetchFromTieredStorage(broker0, topicA, p0, 0)
                .consume(topicA, p0, 0L, 1, 0);
    }

    protected abstract Map<String, String> configsToBeAdded();
}


Review comment (suggestion):
This comment is not very clear to me. How about the following?

`DeleteSegmentsByRetentionTimeTest uses a really small retention time, which could cause the active segment to be rolled and deleted. We use a future timestamp to prevent that from happening.`

Code after:

    @Override
    protected void writeTestSpecifications(TieredStorageTestBuilder builder) {
        final Integer broker0 = 0;
        final String topicA = "topicA";
        final Integer p0 = 0;
        final Integer partitionCount = 1;
        final Integer replicationFactor = 1;
        final Integer maxBatchCountPerSegment = 1;
        final Map<Integer, List<Integer>> replicaAssignment = null;
        final boolean enableRemoteLogStorage = true;
        final int beginEpoch = 0;
        final long startOffset = 3;

        // Create topicA with 1 partition, 1 RF and enabled with remote storage.
        builder.createTopic(topicA, partitionCount, replicationFactor, maxBatchCountPerSegment, replicaAssignment,
                        enableRemoteLogStorage)
                // produce events to partition 0 and expect 3 segments to be offloaded
                .expectSegmentToBeOffloaded(broker0, topicA, p0, 0, new KeyValueSpec("k0", "v0"))
                .expectSegmentToBeOffloaded(broker0, topicA, p0, 1, new KeyValueSpec("k1", "v1"))
                .expectSegmentToBeOffloaded(broker0, topicA, p0, 2, new KeyValueSpec("k2", "v2"))
                .expectEarliestLocalOffsetInLogDirectory(topicA, p0, 3L)
                .produceWithTimestamp(topicA, p0, new KeyValueSpec("k0", "v0"), new KeyValueSpec("k1", "v1"),
                        // DeleteSegmentsByRetentionTimeTest uses a tiny retention time, which could cause the active 
                        // segment to be rolled and deleted. We use a future timestamp to prevent that from happening.
                        new KeyValueSpec("k2", "v2"), new KeyValueSpec("k3", "v3", System.currentTimeMillis() + TimeUnit.HOURS.toMillis(1)))
                // update the topic config such that it triggers the deletion of segments
                .updateTopicConfig(topicA, configsToBeAdded(), Collections.emptyList())
                // expect that the three offloaded remote log segments are deleted
                .expectDeletionInRemoteStorage(broker0, topicA, p0, DELETE_SEGMENT, 3)
                .waitForRemoteLogSegmentDeletion(topicA)
                // expect that the leader epoch checkpoint is updated
                .expectLeaderEpochCheckpoint(broker0, topicA, p0, beginEpoch, startOffset)
                // consume from the beginning of the topic to read data from local and remote storage
                .expectFetchFromTieredStorage(broker0, topicA, p0, 0)
                .consume(topicA, p0, 0L, 1, 0);
    }

    protected abstract Map<String, String> configsToBeAdded();
}