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

feat(s3stream): optimization handler chain #1726

Open
wants to merge 3 commits into
base: main
Choose a base branch
from
Open

feat(s3stream): optimization handler chain #1726

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
42bc519
feat(s3stream): optimization handler chain
CLFutureX Aug 6, 2024
6c47f74
feat(s3stream): optimization handler chain
CLFutureX Aug 6, 2024
bffb264
feat(s3stream): optimization handler chain
CLFutureX Aug 7, 2024
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
9 changes: 5 additions & 4 deletions core/src/main/scala/kafka/log/stream/s3/DefaultS3Client.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import com.automq.stream.s3.failover.ForceCloseStorageFailureHandler;
import com.automq.stream.s3.failover.HaltStorageFailureHandler;
import com.automq.stream.s3.failover.StorageFailureHandlerChain;
import com.automq.stream.s3.failover.StorageHandlerChain;
import com.automq.stream.s3.index.LocalStreamRangeIndexCache;
import com.automq.stream.s3.network.AsyncNetworkBandwidthLimiter;
import com.automq.stream.s3.objects.ObjectManager;
Expand Down Expand Up @@ -125,12 +126,12 @@ public void start() {
this.blockCache = new StreamReaders(this.config.blockCacheSize(), objectManager, objectStorage, objectReaderFactory);
this.compactionManager = new CompactionManager(this.config, this.objectManager, this.streamManager, compactionobjectStorage);
this.writeAheadLog = buildWAL();
StorageFailureHandlerChain storageFailureHandler = new StorageFailureHandlerChain();
this.storage = new S3Storage(this.config, writeAheadLog, streamManager, objectManager, blockCache, objectStorage, storageFailureHandler);
StorageHandlerChain storageHandlerChain = new StorageFailureHandlerChain();
this.storage = new S3Storage(this.config, writeAheadLog, streamManager, objectManager, blockCache, objectStorage, storageHandlerChain);
// stream object compactions share the same object storage with stream set object compactions
this.streamClient = new S3StreamClient(this.streamManager, this.storage, this.objectManager, compactionobjectStorage, this.config, networkInboundLimiter, networkOutboundLimiter);
storageFailureHandler.addHandler(new ForceCloseStorageFailureHandler(streamClient));
storageFailureHandler.addHandler(new HaltStorageFailureHandler());
storageHandlerChain.addHandler(new ForceCloseStorageFailureHandler(streamClient));
storageHandlerChain.addHandler(new HaltStorageFailureHandler());
this.streamClient.registerStreamLifeCycleListener(localIndexCache);
this.kvClient = new ControllerKVClient(this.requestSender);
this.failover = failover();
Expand Down
10 changes: 5 additions & 5 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 @@ -19,7 +19,7 @@
import com.automq.stream.s3.context.AppendContext;
import com.automq.stream.s3.context.FetchContext;
import com.automq.stream.s3.failover.Failover;
import com.automq.stream.s3.failover.StorageFailureHandler;
import com.automq.stream.s3.failover.StorageHandlerChain;
import com.automq.stream.s3.metadata.StreamMetadata;
import com.automq.stream.s3.metrics.S3StreamMetricsManager;
import com.automq.stream.s3.metrics.TimerUtil;
Expand Down Expand Up @@ -112,7 +112,7 @@ public class S3Storage implements Storage {
private final ObjectManager objectManager;
private final ObjectStorage objectStorage;
private final S3BlockCache blockCache;
private final StorageFailureHandler storageFailureHandler;
private final StorageHandlerChain storageHandlerChain;
/**
* Stream callback locks. Used to ensure the stream callbacks will not be called concurrently.
*
Expand All @@ -128,7 +128,7 @@ public class S3Storage implements Storage {

@SuppressWarnings("this-escape")
public S3Storage(Config config, WriteAheadLog deltaWAL, StreamManager streamManager, ObjectManager objectManager,
S3BlockCache blockCache, ObjectStorage objectStorage, StorageFailureHandler storageFailureHandler) {
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why S3Storage need to know the storage failure handle is a StorageHandlerChain?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why S3Storage need to know the storage failure handle is a StorageHandlerChain?

Based on the current constructor method with the handler parameter, it gives the impression that only a single handler implementation class needs to be executed, whereas in reality, S3Storage requires the execution of multiple handler implementation classes.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why S3Storage need to know the storage failure handle is a StorageHandlerChain?

The root cause lies in the fact that the current design does not distinguish between 'chain' and 'handler' (that is, the chain of responsibility has not been distinguished from the nodes on the chain of responsibility), with differentiation made only in naming.

S3BlockCache blockCache, ObjectStorage objectStorage, StorageHandlerChain storageHandlerChain) {
this.config = config;
this.maxDeltaWALCacheSize = config.walCacheSize();
this.deltaWAL = deltaWAL;
Expand All @@ -137,7 +137,7 @@ public S3Storage(Config config, WriteAheadLog deltaWAL, StreamManager streamMana
this.streamManager = streamManager;
this.objectManager = objectManager;
this.objectStorage = objectStorage;
this.storageFailureHandler = storageFailureHandler;
this.storageHandlerChain = storageHandlerChain;
this.drainBackoffTask = this.backgroundExecutor.scheduleWithFixedDelay(this::tryDrainBackoffRecords, 100, 100, TimeUnit.MILLISECONDS);
S3StreamMetricsManager.registerInflightWALUploadTasksCountSupplier(this.inflightWALUploadTasks::size);
S3StreamMetricsManager.registerDeltaWalPendingUploadBytesSupplier(this.pendingUploadBytes::get);
Expand Down Expand Up @@ -468,7 +468,7 @@ public boolean append0(AppendContext context, WalWriteRequest request, boolean f
appendResult.future().whenComplete((nil, ex) -> {
if (ex != null) {
LOGGER.error("append WAL fail, request {}", request, ex);
storageFailureHandler.handle(ex);
storageHandlerChain.handle(ex);
return;
}
handleAppendCallback(request);
Expand Down
3 changes: 2 additions & 1 deletion s3stream/src/main/java/com/automq/stream/s3/failover/StorageFailureHandlerChain.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class StorageFailureHandlerChain implements StorageFailureHandler {
public class StorageFailureHandlerChain implements StorageHandlerChain {
private static final Logger LOGGER = LoggerFactory.getLogger(StorageFailureHandlerChain.class);
private final List<StorageFailureHandler> handlers = new ArrayList<>();

Expand All @@ -31,6 +31,7 @@ public void handle(Throwable ex) {
}
}

@Override
public void addHandler(StorageFailureHandler handler) {
handlers.add(handler);
}
Expand Down
16 changes: 16 additions & 0 deletions s3stream/src/main/java/com/automq/stream/s3/failover/StorageHandlerChain.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,16 @@
/*
* Copyright 2024, AutoMQ HK Limited.
*
* The use of this file is governed by the Business Source License,
* as detailed in the file "/LICENSE.S3Stream" included in this repository.
*
* As of the Change Date specified in that file, in accordance with
* the Business Source License, use of this software will be governed
* by the Apache License, Version 2.0
*/

package com.automq.stream.s3.failover;

public interface StorageHandlerChain extends StorageFailureHandler {
void addHandler(StorageFailureHandler handler);
}
4 changes: 2 additions & 2 deletions s3stream/src/test/java/com/automq/stream/s3/S3StorageTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
import com.automq.stream.s3.cache.ReadDataBlock;
import com.automq.stream.s3.cache.blockcache.DefaultObjectReaderFactory;
import com.automq.stream.s3.cache.blockcache.StreamReaders;
import com.automq.stream.s3.failover.StorageFailureHandler;
import com.automq.stream.s3.failover.StorageHandlerChain;
import com.automq.stream.s3.metadata.StreamMetadata;
import com.automq.stream.s3.metadata.StreamState;
import com.automq.stream.s3.model.StreamRecordBatch;
Expand Down Expand Up @@ -81,7 +81,7 @@ public void setup() {
objectStorage = new MemoryObjectStorage();
storage = new S3Storage(config, wal,
streamManager, objectManager, new StreamReaders(config.blockCacheSize(), objectManager, objectStorage,
new DefaultObjectReaderFactory(objectStorage)), objectStorage, mock(StorageFailureHandler.class));
new DefaultObjectReaderFactory(objectStorage)), objectStorage, mock(StorageHandlerChain.class));
}

@Test
Expand Down
Loading