Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

refactor(s3stream): standardize the code style #864

Merged
merged 3 commits into from
Dec 29, 2023
Merged

refactor(s3stream): standardize the code style #864

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9d405dd
refactor(s3stream): standardize the code style
ShadowySpirits Dec 28, 2023
9c2f1c8
fix(store): fix store
ShadowySpirits Dec 28, 2023
51a29b3
feat(s3stream): optimize error message for checking S3 availability
ShadowySpirits Dec 28, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion s3stream/src/main/java/com/automq/stream/api/CreateStreamOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,10 @@ public class CreateStreamOptions {
private int replicaCount;
private long epoch;

public static Builder newBuilder() {
private CreateStreamOptions() {
}

public static Builder builder() {
return new Builder();
}

Expand Down
5 changes: 4 additions & 1 deletion s3stream/src/main/java/com/automq/stream/api/OpenStreamOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,10 @@ public class OpenStreamOptions {
private ReadMode readMode = ReadMode.MULTIPLE;
private long epoch;

public static Builder newBuilder() {
private OpenStreamOptions() {
}

public static Builder builder() {
return new Builder();
}

Expand Down
6 changes: 3 additions & 3 deletions s3stream/src/main/java/com/automq/stream/s3/S3Storage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -187,13 +187,13 @@ void recover0(WriteAheadLog deltaWAL, StreamManager streamManager, ObjectManager
}
deltaWAL.reset().get();
for (StreamMetadata stream : streams) {
long newEndOffset = streamEndOffsets.getOrDefault(stream.getStreamId(), stream.getEndOffset());
long newEndOffset = streamEndOffsets.getOrDefault(stream.streamId(), stream.endOffset());
logger.info("recover try close stream {} with new end offset {}", stream, newEndOffset);
}
CompletableFuture.allOf(
streams
.stream()
.map(s -> streamManager.closeStream(s.getStreamId(), s.getEpoch()))
.map(s -> streamManager.closeStream(s.streamId(), s.epoch()))
.toArray(CompletableFuture[]::new)
).get();
}
Expand All @@ -203,7 +203,7 @@ static LogCache.LogCacheBlock recoverContinuousRecords(Iterator<WriteAheadLog.Re
}

static LogCache.LogCacheBlock recoverContinuousRecords(Iterator<WriteAheadLog.RecoverResult> it, List<StreamMetadata> openingStreams, Logger logger) {
Map<Long, Long> openingStreamEndOffsets = openingStreams.stream().collect(Collectors.toMap(StreamMetadata::getStreamId, StreamMetadata::getEndOffset));
Map<Long, Long> openingStreamEndOffsets = openingStreams.stream().collect(Collectors.toMap(StreamMetadata::streamId, StreamMetadata::endOffset));
LogCache.LogCacheBlock cacheBlock = new LogCache.LogCacheBlock(1024L * 1024 * 1024);
long logEndOffset = -1L;
Map<Long, Long> streamNextOffsets = new HashMap<>();
Expand Down
9 changes: 4 additions & 5 deletions s3stream/src/main/java/com/automq/stream/s3/S3StreamClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,9 +31,6 @@
import com.automq.stream.utils.FutureUtil;
import com.automq.stream.utils.ThreadUtils;
import com.automq.stream.utils.Threads;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.LinkedList;
import java.util.List;
import java.util.Map;
Expand All @@ -45,6 +42,8 @@
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class S3StreamClient implements StreamClient {
private static final Logger LOGGER = LoggerFactory.getLogger(S3StreamClient.class);
Expand Down Expand Up @@ -145,8 +144,8 @@ private CompletableFuture<Stream> openStream0(long streamId, long epoch) {
.eligibleStreamObjectLivingTimeInMs(config.streamObjectCompactionLivingTimeMinutes() * 60L * 1000)
.s3ObjectLogEnabled(config.objectLogEnable()).executor(streamCompactionExecutor);
S3Stream stream = new S3Stream(
metadata.getStreamId(), metadata.getEpoch(),
metadata.getStartOffset(), metadata.getEndOffset(),
metadata.streamId(), metadata.epoch(),
metadata.startOffset(), metadata.endOffset(),
storage, streamManager, builder, id -> {
openedStreams.remove(id);
return null;
Expand Down
8 changes: 3 additions & 5 deletions s3stream/src/main/java/com/automq/stream/s3/compact/CompactionManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,9 @@

import com.automq.stream.s3.Config;
import com.automq.stream.s3.S3ObjectLogger;
import com.automq.stream.s3.StreamDataBlock;
import com.automq.stream.s3.compact.objects.CompactedObject;
import com.automq.stream.s3.compact.objects.CompactionType;
import com.automq.stream.s3.StreamDataBlock;
import com.automq.stream.s3.compact.operator.DataBlockReader;
import com.automq.stream.s3.compact.operator.DataBlockWriter;
import com.automq.stream.s3.metadata.S3ObjectMetadata;
Expand All @@ -38,7 +38,6 @@
import com.automq.stream.utils.Threads;
import io.github.bucket4j.Bucket;
import io.netty.util.concurrent.DefaultThreadFactory;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
Expand All @@ -56,7 +55,6 @@
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
Expand Down Expand Up @@ -476,7 +474,7 @@ CommitStreamSetObjectRequest buildCompactRequest(List<StreamMetadata> streamMeta
boolean isSanityCheckFailed(List<StreamMetadata> streamMetadataList, List<S3ObjectMetadata> compactedObjects,
CommitStreamSetObjectRequest request) {
Map<Long, StreamMetadata> streamMetadataMap = streamMetadataList.stream()
.collect(Collectors.toMap(StreamMetadata::getStreamId, e -> e));
.collect(Collectors.toMap(StreamMetadata::streamId, e -> e));
Map<Long, S3ObjectMetadata> objectMetadataMap = compactedObjects.stream()
.collect(Collectors.toMap(S3ObjectMetadata::objectId, e -> e));

Expand All @@ -493,7 +491,7 @@ boolean isSanityCheckFailed(List<StreamMetadata> streamMetadataList, List<S3Obje
// skip non-exist stream
continue;
}
long streamStartOffset = streamMetadataMap.get(streamOffsetRange.getStreamId()).getStartOffset();
long streamStartOffset = streamMetadataMap.get(streamOffsetRange.getStreamId()).startOffset();
if (streamOffsetRange.getEndOffset() <= streamStartOffset) {
// skip stream offset range that has been trimmed
continue;
Expand Down
11 changes: 5 additions & 6 deletions s3stream/src/main/java/com/automq/stream/s3/compact/CompactionUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,16 +17,14 @@

package com.automq.stream.s3.compact;

import com.automq.stream.s3.compact.objects.CompactedObjectBuilder;
import com.automq.stream.s3.StreamDataBlock;
import com.automq.stream.s3.compact.objects.CompactedObjectBuilder;
import com.automq.stream.s3.compact.operator.DataBlockReader;
import com.automq.stream.s3.compact.operator.DataBlockWriter;
import com.automq.stream.s3.metadata.S3ObjectMetadata;
import com.automq.stream.s3.metadata.StreamMetadata;
import com.automq.stream.s3.objects.ObjectStreamRange;
import com.automq.stream.s3.operator.S3Operator;
import com.automq.stream.s3.metadata.S3ObjectMetadata;
import org.slf4j.Logger;

import java.util.AbstractMap;
import java.util.ArrayList;
import java.util.HashMap;
Expand All @@ -36,6 +34,7 @@
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;
import org.slf4j.Logger;

public class CompactionUtils {
public static List<ObjectStreamRange> buildObjectStreamRange(List<StreamDataBlock> streamDataBlocks) {
Expand Down Expand Up @@ -73,7 +72,7 @@ public static Map<Long, List<StreamDataBlock>> blockWaitObjectIndices(List<Strea
S3Operator s3Operator,
Logger logger) {
Map<Long, StreamMetadata> streamMetadataMap = streamMetadataList.stream()
.collect(Collectors.toMap(StreamMetadata::getStreamId, s -> s));
.collect(Collectors.toMap(StreamMetadata::streamId, s -> s));
Map<Long, CompletableFuture<List<StreamDataBlock>>> objectStreamRangePositionFutures = new HashMap<>();
for (S3ObjectMetadata objectMetadata : objectMetadataList) {
DataBlockReader dataBlockReader = new DataBlockReader(objectMetadata, s3Operator);
Expand All @@ -91,7 +90,7 @@ public static Map<Long, List<StreamDataBlock>> blockWaitObjectIndices(List<Strea
// non-exist stream
continue;
}
if (streamDataBlock.getEndOffset() <= streamMetadataMap.get(streamDataBlock.getStreamId()).getStartOffset()) {
if (streamDataBlock.getEndOffset() <= streamMetadataMap.get(streamDataBlock.getStreamId()).startOffset()) {
// trimmed stream data block
continue;
}
Expand Down
78 changes: 39 additions & 39 deletions s3stream/src/main/java/com/automq/stream/s3/memory/MemoryMetadataManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,17 +84,17 @@ public synchronized CompletableFuture<CommitStreamSetObjectResponse> commitStrea
assert stream != null;
if (request.getCompactedObjectIds().isEmpty()) {
// Commit new object.
if (stream.getEndOffset() != range.getStartOffset()) {
throw new IllegalArgumentException("stream " + range.getStreamId() + " end offset " + stream.getEndOffset() + " is not equal to start offset of request " + range.getStartOffset());
if (stream.endOffset() != range.getStartOffset()) {
throw new IllegalArgumentException("stream " + range.getStreamId() + " end offset " + stream.endOffset() + " is not equal to start offset of request " + range.getStartOffset());
}
stream.setEndOffset(range.getEndOffset());
stream.endOffset(range.getEndOffset());
} else {
// Compact old object.
if (stream.getEndOffset() < range.getEndOffset()) {
throw new IllegalArgumentException("stream " + range.getStreamId() + " end offset " + stream.getEndOffset() + " is lesser than request " + range.getEndOffset());
if (stream.endOffset() < range.getEndOffset()) {
throw new IllegalArgumentException("stream " + range.getStreamId() + " end offset " + stream.endOffset() + " is lesser than request " + range.getEndOffset());
}
if (stream.getStartOffset() > range.getStartOffset()) {
throw new IllegalArgumentException("stream " + range.getStreamId() + " start offset " + stream.getStartOffset() + " is greater than request " + range.getStartOffset());
if (stream.startOffset() > range.getStartOffset()) {
throw new IllegalArgumentException("stream " + range.getStreamId() + " start offset " + stream.startOffset() + " is greater than request " + range.getStartOffset());
}
}
}
Expand All @@ -111,17 +111,17 @@ public synchronized CompletableFuture<CommitStreamSetObjectResponse> commitStrea
assert stream != null;
if (request.getCompactedObjectIds().isEmpty()) {
// Commit new object.
if (stream.getEndOffset() != streamObject.getStartOffset()) {
throw new IllegalArgumentException("stream " + streamObject.getStreamId() + " end offset " + stream.getEndOffset() + " is not equal to start offset of request " + streamObject.getStartOffset());
if (stream.endOffset() != streamObject.getStartOffset()) {
throw new IllegalArgumentException("stream " + streamObject.getStreamId() + " end offset " + stream.endOffset() + " is not equal to start offset of request " + streamObject.getStartOffset());
}
stream.setEndOffset(streamObject.getEndOffset());
stream.endOffset(streamObject.getEndOffset());
} else {
// Compact old object.
if (stream.getEndOffset() < streamObject.getEndOffset()) {
throw new IllegalArgumentException("stream " + streamObject.getStreamId() + " end offset " + stream.getEndOffset() + " is lesser than request " + streamObject.getEndOffset());
if (stream.endOffset() < streamObject.getEndOffset()) {
throw new IllegalArgumentException("stream " + streamObject.getStreamId() + " end offset " + stream.endOffset() + " is lesser than request " + streamObject.getEndOffset());
}
if (stream.getStartOffset() > streamObject.getStartOffset()) {
throw new IllegalArgumentException("stream " + streamObject.getStreamId() + " start offset " + stream.getStartOffset() + " is greater than request " + streamObject.getStartOffset());
if (stream.startOffset() > streamObject.getStartOffset()) {
throw new IllegalArgumentException("stream " + streamObject.getStreamId() + " start offset " + stream.startOffset() + " is greater than request " + streamObject.getStartOffset());
}
}

Expand All @@ -142,11 +142,11 @@ public synchronized CompletableFuture<Void> compactStreamObject(CompactStreamObj
long streamId = request.getStreamId();
StreamMetadata stream = streams.get(streamId);
assert stream != null;
if (stream.getEndOffset() < request.getEndOffset()) {
throw new IllegalArgumentException("stream " + streamId + " end offset " + stream.getEndOffset() + " is lesser than request " + request.getEndOffset());
if (stream.endOffset() < request.getEndOffset()) {
throw new IllegalArgumentException("stream " + streamId + " end offset " + stream.endOffset() + " is lesser than request " + request.getEndOffset());
}
if (stream.getStartOffset() > request.getStartOffset()) {
throw new IllegalArgumentException("stream " + streamId + " start offset " + stream.getStartOffset() + " is greater than request " + request.getStartOffset());
if (stream.startOffset() > request.getStartOffset()) {
throw new IllegalArgumentException("stream " + streamId + " start offset " + stream.startOffset() + " is greater than request " + request.getStartOffset());
}

streamObjects.computeIfAbsent(streamId, id -> new LinkedList<>())
Expand Down Expand Up @@ -207,7 +207,7 @@ public synchronized CompletableFuture<List<S3ObjectMetadata>> getStreamObjects(l

@Override
public synchronized CompletableFuture<List<StreamMetadata>> getOpeningStreams() {
return CompletableFuture.completedFuture(streams.values().stream().filter(stream -> stream.getState() == StreamState.OPENED).toList());
return CompletableFuture.completedFuture(streams.values().stream().filter(stream -> stream.state() == StreamState.OPENED).toList());
}

@Override
Expand All @@ -228,14 +228,14 @@ public synchronized CompletableFuture<StreamMetadata> openStream(long streamId,
if (stream == null) {
return CompletableFuture.failedFuture(new IllegalArgumentException("stream " + streamId + " not found"));
}
if (stream.getState() == StreamState.OPENED) {
if (stream.state() == StreamState.OPENED) {
return CompletableFuture.failedFuture(new IllegalArgumentException("stream " + streamId + " has been opened"));
}
if (stream.getEpoch() >= epoch) {
return CompletableFuture.failedFuture(new IllegalArgumentException("stream " + streamId + " epoch " + epoch + " is not newer than current epoch " + stream.getEpoch()));
if (stream.epoch() >= epoch) {
return CompletableFuture.failedFuture(new IllegalArgumentException("stream " + streamId + " epoch " + epoch + " is not newer than current epoch " + stream.epoch()));
}
stream.setEpoch(epoch);
stream.setState(StreamState.OPENED);
stream.epoch(epoch);
stream.state(StreamState.OPENED);
return CompletableFuture.completedFuture(stream);
}

Expand All @@ -245,19 +245,19 @@ public synchronized CompletableFuture<Void> trimStream(long streamId, long epoch
if (stream == null) {
return CompletableFuture.failedFuture(new IllegalArgumentException("stream " + streamId + " not found"));
}
if (stream.getState() != StreamState.OPENED) {
if (stream.state() != StreamState.OPENED) {
return CompletableFuture.failedFuture(new IllegalArgumentException("stream " + streamId + " is not opened"));
}
if (stream.getEpoch() != epoch) {
return CompletableFuture.failedFuture(new IllegalArgumentException("stream " + streamId + " epoch " + epoch + " is not equal to current epoch " + stream.getEpoch()));
if (stream.epoch() != epoch) {
return CompletableFuture.failedFuture(new IllegalArgumentException("stream " + streamId + " epoch " + epoch + " is not equal to current epoch " + stream.epoch()));
}
if (newStartOffset < stream.getStartOffset()) {
return CompletableFuture.failedFuture(new IllegalArgumentException("stream " + streamId + " new start offset " + newStartOffset + " is less than current start offset " + stream.getStartOffset()));
if (newStartOffset < stream.startOffset()) {
return CompletableFuture.failedFuture(new IllegalArgumentException("stream " + streamId + " new start offset " + newStartOffset + " is less than current start offset " + stream.startOffset()));
}
if (newStartOffset > stream.getEndOffset()) {
return CompletableFuture.failedFuture(new IllegalArgumentException("stream " + streamId + " new start offset " + newStartOffset + " is greater than current end offset " + stream.getEndOffset()));
if (newStartOffset > stream.endOffset()) {
return CompletableFuture.failedFuture(new IllegalArgumentException("stream " + streamId + " new start offset " + newStartOffset + " is greater than current end offset " + stream.endOffset()));
}
stream.setStartOffset(newStartOffset);
stream.startOffset(newStartOffset);
return CompletableFuture.completedFuture(null);
}

Expand All @@ -267,13 +267,13 @@ public synchronized CompletableFuture<Void> closeStream(long streamId, long epoc
if (stream == null) {
return CompletableFuture.failedFuture(new IllegalArgumentException("stream " + streamId + " not found"));
}
if (stream.getState() != StreamState.OPENED) {
if (stream.state() != StreamState.OPENED) {
return CompletableFuture.failedFuture(new IllegalArgumentException("stream " + streamId + " is not opened"));
}
if (stream.getEpoch() != epoch) {
return CompletableFuture.failedFuture(new IllegalArgumentException("stream " + streamId + " epoch " + epoch + " is not equal to current epoch " + stream.getEpoch()));
if (stream.epoch() != epoch) {
return CompletableFuture.failedFuture(new IllegalArgumentException("stream " + streamId + " epoch " + epoch + " is not equal to current epoch " + stream.epoch()));
}
stream.setState(StreamState.CLOSED);
stream.state(StreamState.CLOSED);
return CompletableFuture.completedFuture(null);
}

Expand All @@ -283,11 +283,11 @@ public synchronized CompletableFuture<Void> deleteStream(long streamId, long epo
if (stream == null) {
return CompletableFuture.failedFuture(new IllegalArgumentException("stream " + streamId + " not found"));
}
if (stream.getState() != StreamState.CLOSED) {
if (stream.state() != StreamState.CLOSED) {
return CompletableFuture.failedFuture(new IllegalArgumentException("stream " + streamId + " is not closed"));
}
if (stream.getEpoch() != epoch) {
return CompletableFuture.failedFuture(new IllegalArgumentException("stream " + streamId + " epoch " + epoch + " is not equal to current epoch " + stream.getEpoch()));
if (stream.epoch() != epoch) {
return CompletableFuture.failedFuture(new IllegalArgumentException("stream " + streamId + " epoch " + epoch + " is not equal to current epoch " + stream.epoch()));
}
streams.remove(streamId);
return CompletableFuture.completedFuture(null);
Expand Down
Loading
Loading