chore(build): add e2e to ci workflow

Aiven-Open · Aug 10, 2023 · 8498791 · 8498791
1 parent fe879c4
commit 8498791
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 10 deletions.
diff --git a/.github/workflows/main_push_and_pull_request_workflow.yml b/.github/workflows/main_push_and_pull_request_workflow.yml
@@ -24,4 +24,8 @@ jobs:
           java-version: ${{ matrix.java-version }}
           distribution: temurin
       - name: Build with Gradle
-        run: ./gradlew build distTar
+        run: ./gradlew build distTar -x e2e:test
+      - name: Build docker image
+        run: make docker_image
+      - name: Run E2E tests
+        run: ./gradlew e2e:test
diff --git a/Makefile b/Makefile
@@ -21,7 +21,7 @@ clean:
 	./gradlew clean
 
 build/distributions/tiered-storage-for-apache-kafka-$(VERSION).tgz:
-	./gradlew build distTar
+	./gradlew build distTar -x e2e:test
 
 .PHONY: docker_image
 docker_image: build/distributions/tiered-storage-for-apache-kafka-$(VERSION).tgz

diff --git a/e2e/src/test/java/io/aiven/kafka/tieredstorage/e2e/internal/RemoteLogMetadataTracker.java b/e2e/src/test/java/io/aiven/kafka/tieredstorage/e2e/internal/RemoteLogMetadataTracker.java
@@ -17,6 +17,7 @@
 package io.aiven.kafka.tieredstorage.e2e.internal;
 
 import java.time.Duration;
+import java.util.ArrayList;
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -122,26 +123,37 @@ private boolean allRemoteSegmentsAreFinished() {
     }
 
     public void waitUntilSegmentsAreDeleted(final List<RemoteSegment> segmentsToBeDeleted) {
-        // TODO set some max wait time
-
         final Supplier<Boolean> segmentsDeleted = () -> segmentsToBeDeleted.stream()
             .allMatch(rs ->
                 remoteSegmentStates.get(rs.remoteLogSegmentId()).equals(RemoteLogSegmentState.DELETE_SEGMENT_FINISHED));
         ConsumerRecords<byte[], RemoteLogMetadata> records;
+
+        final var metadataRecords = new ArrayList<RemoteLogMetadata>();
+        final var timeout = Duration.ofSeconds(30);
+
+        final var startAt = System.currentTimeMillis();
+
         while (!(records = consumer.poll(Duration.ofSeconds(5))).isEmpty() || !segmentsDeleted.get()) {
             records.forEach(r -> {
                 final RemoteLogMetadata metadata = r.value();
+                metadataRecords.add(metadata);
                 if (metadata instanceof RemoteLogSegmentMetadataUpdate) {
-                    final var rlsmu = (RemoteLogSegmentMetadataUpdate) metadata;
-                    remoteSegmentStates.put(rlsmu.remoteLogSegmentId(), rlsmu.state());
-                    if (rlsmu.state().equals(RemoteLogSegmentState.DELETE_SEGMENT_FINISHED)) {
+                    final var metadataUpdate = (RemoteLogSegmentMetadataUpdate) metadata;
+                    remoteSegmentStates.put(metadataUpdate.remoteLogSegmentId(), metadataUpdate.state());
+                    if (metadataUpdate.state().equals(RemoteLogSegmentState.DELETE_SEGMENT_FINISHED)) {
                         remoteSegments.get(metadata.topicIdPartition())
-                            .removeIf(rs -> rs.remoteLogSegmentId().equals(rlsmu.remoteLogSegmentId()));
+                            .removeIf(
+                                segment -> segment.remoteLogSegmentId().equals(metadataUpdate.remoteLogSegmentId()));
                     }
-                } else {
-                    fail("Unexpected metadata: %s", metadata);
                 }
             });
+            if (System.currentTimeMillis() - startAt > timeout.toMillis()) {
+                break;
+            }
+        }
+
+        if (!segmentsDeleted.get()) {
+            fail("Fail to receive delete metadata records. Records received: %s", metadataRecords);
         }
     }
 }