+ * 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 kafka.log.s3; + +import com.automq.elasticstream.client.api.Client; +import com.automq.elasticstream.client.api.KVClient; +import com.automq.elasticstream.client.api.StreamClient; + +public class S3Client implements Client { + @Override + public StreamClient streamClient() { + return null; + } + + @Override + public KVClient kvClient() { + return null; + } +} diff --git a/core/src/main/scala/kafka/log/s3/S3Stream.java b/core/src/main/scala/kafka/log/s3/S3Stream.java new file mode 100644 index 0000000000..8884c89c79 --- /dev/null +++ b/core/src/main/scala/kafka/log/s3/S3Stream.java @@ -0,0 +1,85 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 kafka.log.s3;
+
+import com.automq.elasticstream.client.api.AppendResult;
+import com.automq.elasticstream.client.api.FetchResult;
+import com.automq.elasticstream.client.api.RecordBatch;
+import com.automq.elasticstream.client.api.Stream;
+import kafka.log.s3.cache.S3BlockCache;
+import kafka.log.s3.model.StreamMetadata;
+import kafka.log.s3.objects.ObjectManager;
+import kafka.log.s3.streams.StreamManager;
+
+import java.util.concurrent.CompletableFuture;
+
+public class S3Stream implements Stream {
+ private final StreamMetadata metadata;
+ private final Wal wal;
+ private final S3BlockCache blockCache;
+ private final StreamManager streamManager;
+ private final ObjectManager objectManager;
+
+ public S3Stream(StreamMetadata metadata, Wal wal, S3BlockCache blockCache, StreamManager streamManager, ObjectManager objectManager) {
+ this.metadata = metadata;
+ this.wal = wal;
+ this.blockCache = blockCache;
+ this.streamManager = streamManager;
+ this.objectManager = objectManager;
+ }
+
+ @Override
+ public long streamId() {
+ return metadata.getStreamId();
+ }
+
+ @Override
+ public long startOffset() {
+ return metadata.getStartOffset();
+ }
+
+ @Override
+ public long nextOffset() {
+ return 0;
+ }
+
+ @Override
+ public CompletableFuture
+ * 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 kafka.log.s3;
+
+import com.automq.elasticstream.client.api.CreateStreamOptions;
+import com.automq.elasticstream.client.api.OpenStreamOptions;
+import com.automq.elasticstream.client.api.Stream;
+import com.automq.elasticstream.client.api.StreamClient;
+import kafka.log.s3.cache.S3BlockCache;
+import kafka.log.s3.objects.ObjectManager;
+import kafka.log.s3.streams.StreamManager;
+
+import java.util.concurrent.CompletableFuture;
+
+public class S3StreamClient implements StreamClient {
+ private final StreamManager streamController;
+ private final Wal wal;
+ private final S3BlockCache blockCache;
+ private final ObjectManager objectManager;
+
+ public S3StreamClient(StreamManager streamController, Wal wal, S3BlockCache blockCache, ObjectManager objectManager) {
+ this.streamController = streamController;
+ this.wal = wal;
+ this.blockCache = blockCache;
+ this.objectManager = objectManager;
+ }
+
+ @Override
+ public CompletableFuture
+ * 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 kafka.log.s3;
+
+import kafka.log.s3.model.StreamRecordBatch;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Write ahead log for server.
+ */
+public interface Wal {
+
+ /**
+ * Append stream record to wal.
+ *
+ * @param streamRecord {@link StreamRecordBatch}
+ */
+ CompletableFuture
+ * 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 kafka.log.s3.cache;
+
+import com.automq.elasticstream.client.api.RecordBatch;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Like linux page cache, S3BlockCache is responsible for:
+ * 1. read from S3 when the data block is not in cache.
+ * 2. caching the data blocks of S3 objects.
+ */
+public interface S3BlockCache {
+
+ CompletableFuture
+ * 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 kafka.log.s3.model;
+
+import java.util.OptionalLong;
+
+public class RangeMetadata {
+ private static final long NOOP_OFFSET = -1;
+ private int index;
+ private long startOffset;
+ private long endOffset;
+ private long serverId;
+
+ public int getIndex() {
+ return index;
+ }
+
+ public void setIndex(int index) {
+ this.index = index;
+ }
+
+ public long getStartOffset() {
+ return startOffset;
+ }
+
+ public void setStartOffset(long startOffset) {
+ this.startOffset = startOffset;
+ }
+
+ public OptionalLong getEndOffset() {
+ if (endOffset == NOOP_OFFSET) {
+ return OptionalLong.empty();
+ } else {
+ return OptionalLong.of(endOffset);
+ }
+ }
+
+ public void setEndOffset(long endOffset) {
+ this.endOffset = endOffset;
+ }
+
+ public long getServerId() {
+ return serverId;
+ }
+
+ public void setServerId(long serverId) {
+ this.serverId = serverId;
+ }
+}
diff --git a/core/src/main/scala/kafka/log/s3/model/StreamMetadata.java b/core/src/main/scala/kafka/log/s3/model/StreamMetadata.java
new file mode 100644
index 0000000000..7e43c037d4
--- /dev/null
+++ b/core/src/main/scala/kafka/log/s3/model/StreamMetadata.java
@@ -0,0 +1,60 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kafka.log.s3.model;
+
+import java.util.List;
+
+public class StreamMetadata {
+ private long streamId;
+ private long epoch;
+ private long startOffset;
+
+ private List
+ * 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 kafka.log.s3.model;
+
+import com.automq.elasticstream.client.api.RecordBatch;
+
+public class StreamRecordBatch {
+ private long streamId;
+
+ private RecordBatch recordBatch;
+ private long startOffset;
+}
diff --git a/core/src/main/scala/kafka/log/s3/objects/CommitCompactObjectRequest.java b/core/src/main/scala/kafka/log/s3/objects/CommitCompactObjectRequest.java
new file mode 100644
index 0000000000..be886e636e
--- /dev/null
+++ b/core/src/main/scala/kafka/log/s3/objects/CommitCompactObjectRequest.java
@@ -0,0 +1,59 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kafka.log.s3.objects;
+
+import java.util.List;
+
+public class CommitCompactObjectRequest {
+ private long objectId;
+ private long objectSize;
+ private List
+ * 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 kafka.log.s3.objects;
+
+public class CommitStreamObjectRequest {
+ private long objectId;
+ private long objectSize;
+ private long streamId;
+ private long startOffset;
+ private long endOffset;
+ private long sourceObjectId;
+
+ public long getObjectId() {
+ return objectId;
+ }
+
+ public void setObjectId(long objectId) {
+ this.objectId = objectId;
+ }
+
+ public long getObjectSize() {
+ return objectSize;
+ }
+
+ public void setObjectSize(long objectSize) {
+ this.objectSize = objectSize;
+ }
+
+ public long getStreamId() {
+ return streamId;
+ }
+
+ public void setStreamId(long streamId) {
+ this.streamId = streamId;
+ }
+
+ public long getStartOffset() {
+ return startOffset;
+ }
+
+ public void setStartOffset(long startOffset) {
+ this.startOffset = startOffset;
+ }
+
+ public long getEndOffset() {
+ return endOffset;
+ }
+
+ public void setEndOffset(long endOffset) {
+ this.endOffset = endOffset;
+ }
+
+ public long getSourceObjectId() {
+ return sourceObjectId;
+ }
+
+ public void setSourceObjectId(long sourceObjectId) {
+ this.sourceObjectId = sourceObjectId;
+ }
+}
diff --git a/core/src/main/scala/kafka/log/s3/objects/CommitWalObjectRequest.java b/core/src/main/scala/kafka/log/s3/objects/CommitWalObjectRequest.java
new file mode 100644
index 0000000000..d4d12ee66c
--- /dev/null
+++ b/core/src/main/scala/kafka/log/s3/objects/CommitWalObjectRequest.java
@@ -0,0 +1,50 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kafka.log.s3.objects;
+
+import java.util.List;
+
+public class CommitWalObjectRequest {
+ private long objectId;
+ private long objectSize;
+ private List
+ * 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 kafka.log.s3.objects;
+
+import java.util.List;
+
+/**
+ * Commit wal loose object response.
+ * When stream is fenced, the stream id will be added to failedStreamIds.
+ */
+public class CommitWalObjectResponse {
+ private List
+ * 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 kafka.log.s3.objects;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Object metadata registry.
+ */
+public interface ObjectManager {
+
+ /**
+ * Prepare object id for write, if the objects is not committed in ttl, then delete it.
+ *
+ * @param count object id count.
+ * @param ttl ttl in milliseconds.
+ * @return object id range start.
+ */
+ CompletableFuture
+ * 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 kafka.log.s3.objects;
+
+public class ObjectRange {
+ private long objectId;
+ private long startPosition;
+ private long endPosition;
+
+ public ObjectRange(long objectId, long startPosition, long endPosition) {
+ this.objectId = objectId;
+ this.startPosition = startPosition;
+ this.endPosition = endPosition;
+ }
+
+ public long getObjectId() {
+ return objectId;
+ }
+
+ public void setObjectId(long objectId) {
+ this.objectId = objectId;
+ }
+
+ public long getStartPosition() {
+ return startPosition;
+ }
+
+ public void setStartPosition(long startPosition) {
+ this.startPosition = startPosition;
+ }
+
+ public long getEndPosition() {
+ return endPosition;
+ }
+
+ public void setEndPosition(long endPosition) {
+ this.endPosition = endPosition;
+ }
+}
diff --git a/core/src/main/scala/kafka/log/s3/objects/StreamObject.java b/core/src/main/scala/kafka/log/s3/objects/StreamObject.java
new file mode 100644
index 0000000000..6de0400d90
--- /dev/null
+++ b/core/src/main/scala/kafka/log/s3/objects/StreamObject.java
@@ -0,0 +1,66 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kafka.log.s3.objects;
+
+public class StreamObject {
+ private long objectId;
+ private long objectSize;
+ private long streamId;
+ private long startOffset;
+ private long endOffset;
+
+ public long getObjectId() {
+ return objectId;
+ }
+
+ public void setObjectId(long objectId) {
+ this.objectId = objectId;
+ }
+
+ public long getObjectSize() {
+ return objectSize;
+ }
+
+ public void setObjectSize(long objectSize) {
+ this.objectSize = objectSize;
+ }
+
+ public long getStreamId() {
+ return streamId;
+ }
+
+ public void setStreamId(long streamId) {
+ this.streamId = streamId;
+ }
+
+ public long getStartOffset() {
+ return startOffset;
+ }
+
+ public void setStartOffset(long startOffset) {
+ this.startOffset = startOffset;
+ }
+
+ public long getEndOffset() {
+ return endOffset;
+ }
+
+ public void setEndOffset(long endOffset) {
+ this.endOffset = endOffset;
+ }
+}
diff --git a/core/src/main/scala/kafka/log/s3/objects/WalObjectStreamIndex.java b/core/src/main/scala/kafka/log/s3/objects/WalObjectStreamIndex.java
new file mode 100644
index 0000000000..5c149eda03
--- /dev/null
+++ b/core/src/main/scala/kafka/log/s3/objects/WalObjectStreamIndex.java
@@ -0,0 +1,48 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kafka.log.s3.objects;
+
+public class WalObjectStreamIndex {
+ private long streamId;
+ private long startOffset;
+ private long endOffset;
+
+ public long getStreamId() {
+ return streamId;
+ }
+
+ public void setStreamId(long streamId) {
+ this.streamId = streamId;
+ }
+
+ public long getStartOffset() {
+ return startOffset;
+ }
+
+ public void setStartOffset(long startOffset) {
+ this.startOffset = startOffset;
+ }
+
+ public long getEndOffset() {
+ return endOffset;
+ }
+
+ public void setEndOffset(long endOffset) {
+ this.endOffset = endOffset;
+ }
+}
diff --git a/core/src/main/scala/kafka/log/s3/operator/S3Operator.java b/core/src/main/scala/kafka/log/s3/operator/S3Operator.java
new file mode 100644
index 0000000000..9a9a6a1b1e
--- /dev/null
+++ b/core/src/main/scala/kafka/log/s3/operator/S3Operator.java
@@ -0,0 +1,61 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 kafka.log.s3.operator;
+
+import io.netty.buffer.ByteBuf;
+
+import java.util.concurrent.CompletableFuture;
+
+public interface S3Operator {
+
+ /**
+ * Read data from object.
+ *
+ * @param path object path.
+ * @return data.
+ */
+ CompletableFuture
+ * 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 kafka.log.s3.operator;
+
+import io.netty.buffer.ByteBuf;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Multi-part object writer.
+ */
+public interface Writer {
+ /**
+ * Write a part of the object. The parts will parallel upload to S3.
+ *
+ * @param part object part.
+ */
+ void write(ByteBuf part);
+
+ /**
+ * Copy a part of the object.
+ *
+ * @param sourcePath source object path.
+ * @param start start position of the source object.
+ * @param end end position of the source object.
+ */
+ void copyWrite(String sourcePath, long start, long end);
+
+ /**
+ * Complete the object.
+ */
+ CompletableFuture
+ * 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 kafka.log.s3.streams;
+
+import kafka.log.s3.model.StreamMetadata;
+
+import java.util.concurrent.CompletableFuture;
+
+public interface StreamManager {
+
+ /**
+ * Create a new stream.
+ *
+ * @return stream id.
+ */
+ CompletableFuture