remotePartitionFiles(final TopicIdPartition topicIdPartition);
+
+ abstract boolean assertNoTopicDataOnTierStorage(final String topicName, final Uuid topicId);
+
+ /**
+ * Flag when a step has failed so next steps fail-fast.
+ *
+ * Needed to allow running {@code SingleBrokerTest#stopKafka} after all tests.
+ */
+ public static class FailFastExtension implements TestWatcher, ExecutionCondition {
+ private boolean failed;
+
+ @Override
+ public void testFailed(final ExtensionContext context, final Throwable cause) {
+ LOG.error("Test failed: " + context.getDisplayName());
+ failed = true;
+ }
+
+ @Override
+ public ConditionEvaluationResult evaluateExecutionCondition(final ExtensionContext extensionContext) {
+ if (failed) {
+ return ConditionEvaluationResult.disabled("Already failed");
+ } else {
+ return ConditionEvaluationResult.enabled("Continue testing");
+ }
+ }
+ }
+}
diff --git a/e2e/src/test/java/io/aiven/kafka/tieredstorage/e2e/internal/RemoteLogMetadataDeserializer.java b/e2e/src/test/java/io/aiven/kafka/tieredstorage/e2e/internal/RemoteLogMetadataDeserializer.java
new file mode 100644
index 000000000..232099762
--- /dev/null
+++ b/e2e/src/test/java/io/aiven/kafka/tieredstorage/e2e/internal/RemoteLogMetadataDeserializer.java
@@ -0,0 +1,30 @@
+/*
+ * Copyright 2023 Aiven Oy
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.aiven.kafka.tieredstorage.e2e.internal;
+
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.server.log.remote.metadata.storage.serialization.RemoteLogMetadataSerde;
+import org.apache.kafka.server.log.remote.storage.RemoteLogMetadata;
+
+public class RemoteLogMetadataDeserializer implements Deserializer {
+ private final RemoteLogMetadataSerde serde = new RemoteLogMetadataSerde();
+
+ @Override
+ public RemoteLogMetadata deserialize(final String topic, final byte[] data) {
+ return serde.deserialize(data);
+ }
+}
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
new file mode 100644
index 000000000..76ed8cff3
--- /dev/null
+++ b/e2e/src/test/java/io/aiven/kafka/tieredstorage/e2e/internal/RemoteLogMetadataTracker.java
@@ -0,0 +1,233 @@
+/*
+ * Copyright 2023 Aiven Oy
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.aiven.kafka.tieredstorage.e2e.internal;
+
+import java.time.Duration;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.server.log.remote.storage.RemoteLogMetadata;
+import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentId;
+import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadata;
+import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadataUpdate;
+import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentState;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
+import static org.awaitility.Awaitility.await;
+
+public class RemoteLogMetadataTracker {
+ private static final Logger LOG = LoggerFactory.getLogger(RemoteLogMetadataTracker.class);
+ private static final String REMOTE_LOG_METADATA_TOPIC = "__remote_log_metadata";
+
+ private final KafkaConsumer consumer;
+ private List partitions;
+
+ private final Map> remoteSegments = new HashMap<>();
+ private final Map remoteSegmentStates = new HashMap<>();
+
+ public RemoteLogMetadataTracker(final String bootstrapServers) {
+ consumer = new KafkaConsumer<>(Map.of(
+ "bootstrap.servers", bootstrapServers,
+ "auto.offset.reset", "earliest"
+ ), new ByteArrayDeserializer(), new RemoteLogMetadataDeserializer());
+ }
+
+ public Map> remoteSegments() {
+ final Map> result = new HashMap<>();
+ for (final Map.Entry> entry : remoteSegments.entrySet()) {
+ final List list = entry.getValue().stream()
+ .sorted(Comparator.comparing(RemoteSegment::startOffset))
+ .collect(Collectors.toList());
+ result.put(entry.getKey(), list);
+ }
+ return result;
+ }
+
+ /**
+ * Initializes the tracker.
+ *
+ * It expects at least one record to be present in __remote_log_metadata
+ * and that all remote segments are in {@code COPY_SEGMENT_FINISHED} state.
+ */
+ public void initialize(final List expectedPartitions) {
+ partitions = consumer.partitionsFor(REMOTE_LOG_METADATA_TOPIC).stream()
+ .map(pi -> new TopicPartition(pi.topic(), pi.partition()))
+ .collect(Collectors.toList());
+
+ await().atMost(Duration.ofSeconds(60))
+ .pollInterval(Duration.ofMillis(100))
+ .until(() ->
+ consumer.endOffsets(partitions).values().stream()
+ .mapToLong(Long::longValue)
+ .sum() > 0);
+
+ // supply segment states where copy has not finished
+ final Supplier> segmentsStillCopying = () ->
+ remoteSegmentStates.entrySet().stream()
+ .filter(s -> s.getValue() != RemoteLogSegmentState.COPY_SEGMENT_FINISHED)
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+ consumer.assign(partitions);
+ consumer.seekToBeginning(partitions);
+
+ LOG.info("Initializing remote segments");
+
+ final var timeout = Duration.ofMinutes(1);
+ final var startAt = System.currentTimeMillis();
+
+ final var metadataRecords = new LinkedHashMap, RemoteLogMetadata>();
+ boolean isReady = false;
+ while (!isReady) {
+ consumer.poll(Duration.ofSeconds(5)).forEach(r -> {
+ final RemoteLogMetadata metadata = r.value();
+ assertThat(metadata)
+ .withFailMessage("Unexpected metadata type: %s", metadata)
+ .isInstanceOfAny(RemoteLogSegmentMetadata.class, RemoteLogSegmentMetadataUpdate.class);
+
+ remoteSegments.putIfAbsent(metadata.topicIdPartition(), new HashSet<>());
+ metadataRecords.put(Map.entry(new TopicPartition(r.topic(), r.partition()), r.offset()), metadata);
+ if (metadata instanceof RemoteLogSegmentMetadata) {
+ final var segmentMetadata = (RemoteLogSegmentMetadata) metadata;
+ remoteSegmentStates.put(segmentMetadata.remoteLogSegmentId(), segmentMetadata.state());
+ remoteSegments.get(metadata.topicIdPartition()).add(
+ new RemoteSegment(
+ segmentMetadata.remoteLogSegmentId(),
+ segmentMetadata.startOffset(),
+ segmentMetadata.endOffset()
+ ));
+ } else if (metadata instanceof RemoteLogSegmentMetadataUpdate) {
+ final var update = (RemoteLogSegmentMetadataUpdate) metadata;
+ remoteSegmentStates.put(update.remoteLogSegmentId(), update.state());
+
+ // Sanity check: if we see an update, the original record should be already taken into account.
+ assertThat(remoteSegments.get(metadata.topicIdPartition()))
+ .map(RemoteSegment::remoteLogSegmentId)
+ .contains(update.remoteLogSegmentId());
+ }
+ });
+
+ isReady = segmentsStillCopying.get().isEmpty() // copies have not finished
+ && expectedPartitions.stream().allMatch(remoteSegments::containsKey); // AND not all segments present
+
+ // check for timeout
+ final var running = System.currentTimeMillis() - startAt;
+ if (running > timeout.toMillis()) {
+ LOG.warn("Timeout waiting for segments copy finished events to arrive after {} running", running);
+ break;
+ }
+ }
+
+ if (!isReady) { // if finished because of timeout
+ final var notCopied = segmentsStillCopying.get();
+ fail("Fail to receive copy metadata records for %s out of %s segments."
+ + "%nSegments missing: %n%s"
+ + "%nMetadata events received: %n%s",
+ notCopied.size(), remoteSegments.size(),
+ notCopied.entrySet().stream()
+ .map(e -> e.getKey().toString()
+ + " => "
+ + e.getValue())
+ .collect(Collectors.joining("\n")),
+ metadataRecords.entrySet().stream()
+ .map(e -> e.getKey().getKey()
+ + "-" + e.getKey().getValue() + ":" + e.getValue() + " => "
+ + e.getValue())
+ .collect(Collectors.joining("\n"))
+ );
+ }
+
+ assertThat(remoteSegments.keySet()).containsExactlyInAnyOrderElementsOf(expectedPartitions);
+
+ LOG.info("Remote Log Metadata Tracker initialized");
+ }
+
+ public void waitUntilSegmentsAreDeleted(final List segmentsToBeDeleted) {
+ final Supplier> segmentsNotDeleted =
+ () -> segmentsToBeDeleted.stream()
+ .map(rs -> Map.entry(rs, remoteSegmentStates.get(rs.remoteLogSegmentId())))
+ .filter(rs -> rs.getValue() != RemoteLogSegmentState.DELETE_SEGMENT_FINISHED)
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+ LOG.info("Starting validation of remote segments deleted");
+
+ final var timeout = Duration.ofMinutes(1);
+ final var startAt = System.currentTimeMillis();
+
+ final var metadataRecords = new LinkedHashMap, RemoteLogMetadata>();
+
+ boolean isReady = false;
+ while (!isReady) {
+ (consumer.poll(Duration.ofSeconds(5))).forEach(r -> {
+ final RemoteLogMetadata metadata = r.value();
+ metadataRecords.put(Map.entry(new TopicPartition(r.topic(), r.partition()), r.offset()), metadata);
+ if (metadata instanceof RemoteLogSegmentMetadataUpdate) {
+ final var metadataUpdate = (RemoteLogSegmentMetadataUpdate) metadata;
+ remoteSegmentStates.put(metadataUpdate.remoteLogSegmentId(), metadataUpdate.state());
+ if (metadataUpdate.state() == RemoteLogSegmentState.DELETE_SEGMENT_FINISHED) {
+ remoteSegments.get(metadata.topicIdPartition())
+ .removeIf(
+ segment -> segment.remoteLogSegmentId().equals(metadataUpdate.remoteLogSegmentId()));
+ }
+ }
+ });
+
+ isReady = segmentsNotDeleted.get().isEmpty();
+
+ final var running = System.currentTimeMillis() - startAt;
+ if (running > timeout.toMillis()) {
+ LOG.warn("Timeout waiting for segments delete finished events to arrive after {} running", running);
+ break;
+ }
+ }
+
+ if (!segmentsNotDeleted.get().isEmpty()) {
+ final var notDeleted = segmentsNotDeleted.get();
+ fail("Fail to receive delete metadata records for %s out of %s segments."
+ + "%nSegments missing: %n%s"
+ + "%nMetadata events received: %n%s",
+ notDeleted.size(), segmentsToBeDeleted.size(),
+ notDeleted.entrySet().stream()
+ .map(e -> e.getKey().remoteLogSegmentId().toString()
+ + "[" + e.getKey().startOffset() + "-" + e.getKey().endOffset() + "] => "
+ + e.getValue())
+ .collect(Collectors.joining("\n")),
+ metadataRecords.entrySet().stream()
+ .map(e -> e.getKey().getKey()
+ + "-" + e.getKey().getValue() + ":" + e.getValue() + " => "
+ + e.getValue())
+ .collect(Collectors.joining("\n"))
+ );
+ }
+
+ LOG.info("Remote segments deleted validation complete");
+ }
+}
diff --git a/e2e/src/test/java/io/aiven/kafka/tieredstorage/e2e/internal/RemoteSegment.java b/e2e/src/test/java/io/aiven/kafka/tieredstorage/e2e/internal/RemoteSegment.java
new file mode 100644
index 000000000..aa00b0eae
--- /dev/null
+++ b/e2e/src/test/java/io/aiven/kafka/tieredstorage/e2e/internal/RemoteSegment.java
@@ -0,0 +1,75 @@
+/*
+ * Copyright 2023 Aiven Oy
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.aiven.kafka.tieredstorage.e2e.internal;
+
+import java.util.Objects;
+
+import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentId;
+
+public final class RemoteSegment {
+ private final RemoteLogSegmentId remoteLogSegmentId;
+ private final long startOffset;
+ private final long endOffset;
+
+ RemoteSegment(final RemoteLogSegmentId remoteLogSegmentId,
+ final long startOffset,
+ final long endOffset) {
+ this.remoteLogSegmentId = remoteLogSegmentId;
+ this.startOffset = startOffset;
+ this.endOffset = endOffset;
+ }
+
+ public RemoteLogSegmentId remoteLogSegmentId() {
+ return remoteLogSegmentId;
+ }
+
+ public long startOffset() {
+ return startOffset;
+ }
+
+ public long endOffset() {
+ return endOffset;
+ }
+
+ @Override
+ public boolean equals(final Object obj) {
+ if (obj == this) {
+ return true;
+ }
+ if (obj == null || obj.getClass() != this.getClass()) {
+ return false;
+ }
+ final var that = (RemoteSegment) obj;
+ return Objects.equals(this.remoteLogSegmentId, that.remoteLogSegmentId)
+ && this.startOffset == that.startOffset
+ && this.endOffset == that.endOffset;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(remoteLogSegmentId, startOffset, endOffset);
+ }
+
+ @Override
+ public String toString() {
+ return "RemoteSegment["
+ + "remoteLogSegmentId=" + remoteLogSegmentId + ", "
+ + "startOffset=" + startOffset + ", "
+ + "endOffset=" + endOffset + ']';
+ }
+
+}
diff --git a/e2e/src/test/resources/log4j.properties b/e2e/src/test/resources/log4j.properties
new file mode 100644
index 000000000..9fed516a4
--- /dev/null
+++ b/e2e/src/test/resources/log4j.properties
@@ -0,0 +1,28 @@
+##
+# Copyright 2023 Aiven Oy
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+##
+
+log4j.rootLogger=INFO, stdout
+
+log4j.appender.stdout=org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c)%n
+
+log4j.logger.org.apache.kafka.clients.consumer.KafkaConsumer=WARN
+
+org.testcontainers=INFO
+tc=INFO
+com.github.dockerjava=WARN
+com.github.dockerjava.zerodep.shaded.org.apache.hc.client5.http.wire=OFF
diff --git a/settings.gradle b/settings.gradle
index 841bc9f67..10c928539 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -21,3 +21,4 @@ include 'storage:core'
include 'storage:filesystem'
include 'storage:gcs'
include 'storage:s3'
+include 'e2e'
diff --git a/storage/s3/build.gradle b/storage/s3/build.gradle
index 0db62f656..50001e516 100644
--- a/storage/s3/build.gradle
+++ b/storage/s3/build.gradle
@@ -16,12 +16,6 @@
archivesBaseName = "storage-s3"
-ext {
- // Keep empty lines between versions to avoid conflicts on mass update (e.g. Dependabot).
-
- awsSdkVersion = "1.12.520"
-}
-
dependencies {
implementation project(":storage:core")