updated the integration test with more assertions

Author: kamalcph

storage/src/test/java/org/apache/kafka/tiered/storage/integration/ListOffsetsTest.java

@@ -30,6 +30,7 @@
 
 import static org.apache.kafka.common.utils.Utils.mkEntry;
 import static org.apache.kafka.common.utils.Utils.mkMap;
+import static org.apache.kafka.server.log.remote.storage.LocalTieredStorageEvent.EventType.DELETE_SEGMENT;
 
 public class ListOffsetsTest extends TieredStorageTestHarness {
     @Override
@@ -94,7 +95,17 @@ protected void writeTestSpecifications(TieredStorageTestBuilder builder) {
                 .expectListOffsets(broker1, topicA, p0, OffsetSpec.forTimestamp(timestamp + 3), new EpochEntry(Integer.MAX_VALUE, 3))
 
                 // delete some records and check whether the earliest_offset gets updated.
+                .expectDeletionInRemoteStorage(broker1, topicA, p0, DELETE_SEGMENT, 1)
                 .deleteRecords(topicA, p0, 3L)
-                .expectListOffsets(broker1, topicA, p0, OffsetSpec.earliest(), new EpochEntry(Integer.MAX_VALUE, 3));
+                .expectListOffsets(broker1, topicA, p0, OffsetSpec.earliest(), new EpochEntry(Integer.MAX_VALUE, 3))
+                .expectListOffsets(broker1, topicA, p0, OffsetSpec.earliestLocalSpec(), new EpochEntry(Integer.MAX_VALUE, 4))
+
+                // delete all the records in remote layer and expect that earliest and earliest_local offsets are same
+                .expectDeletionInRemoteStorage(broker1, topicA, p0, DELETE_SEGMENT, 1)
+                .deleteRecords(topicA, p0, 4L)
+                .expectListOffsets(broker1, topicA, p0, OffsetSpec.earliest(), new EpochEntry(Integer.MAX_VALUE, 4))
+                .expectListOffsets(broker1, topicA, p0, OffsetSpec.earliestLocalSpec(), new EpochEntry(Integer.MAX_VALUE, 4))
+                .expectListOffsets(broker1, topicA, p0, OffsetSpec.latestTierSpec(), new EpochEntry(-1, 3))
+                .expectListOffsets(broker1, topicA, p0, OffsetSpec.latest(), new EpochEntry(Integer.MAX_VALUE, 6));
     }
 }