Skip to content

Commit

Permalink
refactor(s3stream): standardize the code style (#864)
Browse files Browse the repository at this point in the history 
Signed-off-by: SSpirits <[email protected]>
  • Loading branch information
@ShadowySpirits
ShadowySpirits authored Dec 29, 2023
1 parent 1a1f5ba commit f4e37d1
Show file tree
Hide file tree
Showing 13 changed files with 111 additions and 103 deletions.
5 changes: 4 additions & 1 deletion s3stream/src/main/java/com/automq/stream/api/CreateStreamOptions.java
View file
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
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
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
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
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
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
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

0 comments on commit f4e37d1

Please sign in to comment.