fix(s3stream): Fix "this escape" issues in constructors(#2442) (#2463)

* Fix "this escape" issues in constructors

* DefaultS3Client fix this escape

* confilct with the main branch is solved used factory method in LocalStreamRangeIndexCacheTest.java

Author: Vist233

core/src/main/scala/kafka/log/stream/s3/DefaultS3Client.java

@@ -139,7 +139,7 @@ public void start() {
             config.networkBaselineBandwidth() - (long) networkOutboundRate.derive(
                 TimeUnit.NANOSECONDS.toSeconds(System.nanoTime()), NetworkStats.getInstance().networkOutboundUsageTotal().get()));
 
-        this.localIndexCache = new LocalStreamRangeIndexCache();
+        this.localIndexCache = LocalStreamRangeIndexCache.create();
         this.objectReaderFactory = new DefaultObjectReaderFactory(() -> this.mainObjectStorage);
         this.metadataManager = new StreamMetadataManager(brokerServer, config.nodeId(), objectReaderFactory, localIndexCache);
         this.requestSender = new ControllerRequestSender(brokerServer, new ControllerRequestSender.RetryPolicyContext(config.controllerRequestRetryMaxCount(),

s3stream/src/main/java/com/automq/stream/s3/S3Stream.java

@@ -100,12 +100,7 @@ public class S3Stream implements Stream, StreamMetadataListener {
     private CompletableFuture<Void> closeCf;
     private StreamMetadataListener.Handle listenerHandle;
 
-    public S3Stream(long streamId, long epoch, long startOffset, long nextOffset, Storage storage,
-        StreamManager streamManager) {
-        this(streamId, epoch, startOffset, nextOffset, storage, streamManager, null, null, OpenStreamOptions.DEFAULT);
-    }
-
-    public S3Stream(long streamId, long epoch, long startOffset, long nextOffset, Storage storage,
+    private S3Stream(long streamId, long epoch, long startOffset, long nextOffset, Storage storage,
         StreamManager streamManager, NetworkBandwidthLimiter networkInboundLimiter,
         NetworkBandwidthLimiter networkOutboundLimiter, OpenStreamOptions options) {
         this.streamId = streamId;
@@ -121,6 +116,22 @@ public S3Stream(long streamId, long epoch, long startOffset, long nextOffset, St
         this.networkInboundLimiter = networkInboundLimiter;
         this.networkOutboundLimiter = networkOutboundLimiter;
         this.options = options;
+    }
+
+    public static S3Stream create(long streamId, long epoch, long startOffset, long nextOffset, Storage storage,
+        StreamManager streamManager) {
+        return create(streamId, epoch, startOffset, nextOffset, storage, streamManager, null, null, OpenStreamOptions.DEFAULT);
+    }
+
+    public static S3Stream create(long streamId, long epoch, long startOffset, long nextOffset, Storage storage,
+        StreamManager streamManager, NetworkBandwidthLimiter networkInboundLimiter,
+        NetworkBandwidthLimiter networkOutboundLimiter, OpenStreamOptions options) {
+        S3Stream s3Stream = new S3Stream(streamId, epoch, startOffset, nextOffset, storage, streamManager, networkInboundLimiter, networkOutboundLimiter, options);
+        s3Stream.completeInitialization();
+        return s3Stream;
+    }
+
+    private void completeInitialization() {
         if (snapshotRead()) {
             listenerHandle = streamManager.addMetadataListener(streamId, this);
         }

s3stream/src/main/java/com/automq/stream/s3/S3StreamClient.java

@@ -203,7 +203,7 @@ private CompletableFuture<Stream> openStream0(long streamId, long epoch, Map<Str
     }
 
     S3Stream newStream(StreamMetadata metadata, OpenStreamOptions options) {
-        return new S3Stream(
+        return S3Stream.create(
             metadata.streamId(), metadata.epoch(),
             metadata.startOffset(), metadata.endOffset(),
             storage, streamManager, networkInboundBucket, networkOutboundBucket, options);

s3stream/src/main/java/com/automq/stream/s3/cache/AsyncLRUCache.java

@@ -48,13 +48,21 @@ public class AsyncLRUCache<K, V extends AsyncMeasurable> {
     final Set<V> completedSet = new HashSet<>();
     final Set<V> removedSet = new HashSet<>();
 
-    public AsyncLRUCache(String cacheName, long maxSize) {
+    protected AsyncLRUCache(String cacheName, long maxSize) {
         this.cacheName = cacheName;
         if (maxSize <= 0) {
             throw new IllegalArgumentException("maxSize must be positive");
         }
         this.maxSize = maxSize;
+    }
+
+    public static <K, V extends AsyncMeasurable> AsyncLRUCache<K, V> create(String cacheName, long maxSize) {
+        AsyncLRUCache<K, V> asyncLRUCache = new AsyncLRUCache<>(cacheName, maxSize);
+        asyncLRUCache.completeInitialization();
+        return asyncLRUCache;
+    }
 
+    private void completeInitialization() {
         S3StreamMetricsManager.registerAsyncCacheSizeSupplier(this::totalSize, cacheName);
         S3StreamMetricsManager.registerAsyncCacheMaxSizeSupplier(() -> maxSize, cacheName);
         S3StreamMetricsManager.registerAsyncCacheItemNumberSupplier(this::size, cacheName);

s3stream/src/main/java/com/automq/stream/s3/exceptions/ObjectNotExistException.java

@@ -29,8 +29,7 @@ public ObjectNotExistException(long objectId) {
     }
 
     public ObjectNotExistException(Throwable cause) {
-        super(cause.getMessage());
-        this.addSuppressed(cause);
+        super(cause.getMessage(), cause);
     }
 
 }

s3stream/src/main/java/com/automq/stream/s3/index/LocalStreamRangeIndexCache.java

@@ -80,7 +80,17 @@ public class LocalStreamRangeIndexCache implements S3StreamClient.StreamLifeCycl
     private CompletableFuture<Void> uploadCf = CompletableFuture.completedFuture(null);
     private long lastUploadTime = 0L;
 
-    public LocalStreamRangeIndexCache() {
+    private LocalStreamRangeIndexCache() {
+        
+    }
+
+    public static LocalStreamRangeIndexCache create() {
+        LocalStreamRangeIndexCache cache = new LocalStreamRangeIndexCache();
+        cache.completeInitialization();
+        return cache;
+    }
+
+    private void completeInitialization() {
         S3StreamMetricsManager.registerLocalStreamRangeIndexCacheSizeSupplier(this::totalSize);
         S3StreamMetricsManager.registerLocalStreamRangeIndexCacheStreamNumSupplier(() -> {
             readLock.lock();

s3stream/src/test/java/com/automq/stream/s3/S3StreamTest.java

@@ -52,7 +52,7 @@ public class S3StreamTest {
     public void setup() {
         storage = mock(Storage.class);
         streamManager = mock(StreamManager.class);
-        stream = new S3Stream(233, 1, 100, 233, storage, streamManager);
+        stream = S3Stream.create(233L, 1L, 100L, 233L, storage, streamManager);
     }
 
     @Test

s3stream/src/test/java/com/automq/stream/s3/index/LocalStreamRangeIndexCacheTest.java

@@ -44,7 +44,7 @@ public class LocalStreamRangeIndexCacheTest {
     public void testInit() {
         ObjectStorage objectStorage = new MemoryObjectStorage();
         // init with empty index
-        LocalStreamRangeIndexCache cache = new LocalStreamRangeIndexCache();
+        LocalStreamRangeIndexCache cache = LocalStreamRangeIndexCache.create();
         cache.start();
         cache.init(NODE_0, objectStorage);
         Assertions.assertEquals(-1, cache.searchObjectId(STREAM_0, 0).join());
@@ -55,7 +55,7 @@ public void testInit() {
         cache.updateIndexFromRequest(request);
         cache.upload().join();
 
-        cache = new LocalStreamRangeIndexCache();
+        cache = LocalStreamRangeIndexCache.create();
         cache.start();
         cache.init(NODE_0, objectStorage);
         cache.initCf().join();
@@ -69,7 +69,7 @@ public void testInit() {
     @Test
     public void testAppend() {
         ObjectStorage objectStorage = new MemoryObjectStorage();
-        LocalStreamRangeIndexCache cache = new LocalStreamRangeIndexCache();
+        LocalStreamRangeIndexCache cache = LocalStreamRangeIndexCache.create();
         cache.start();
         cache.init(NODE_0, objectStorage);
         CommitStreamSetObjectRequest request = new CommitStreamSetObjectRequest();
@@ -93,7 +93,7 @@ public void testAppend() {
     @Test
     public void testPrune() {
         ObjectStorage objectStorage = new MemoryObjectStorage();
-        LocalStreamRangeIndexCache cache = new LocalStreamRangeIndexCache();
+        LocalStreamRangeIndexCache cache = LocalStreamRangeIndexCache.create();
         cache.start();
         cache.init(NODE_0, objectStorage);
         CommitStreamSetObjectRequest request = new CommitStreamSetObjectRequest();
@@ -130,7 +130,7 @@ public void testPrune() {
         Assertions.assertEquals(97, cache.searchObjectId(STREAM_0, 1500).join());
 
         // test load from object storage
-        cache = new LocalStreamRangeIndexCache();
+        cache = LocalStreamRangeIndexCache.create();
         cache.start();
         cache.init(NODE_0, objectStorage);
         cache.initCf().join();
@@ -155,7 +155,7 @@ public void testPrune() {
     @Test
     public void testEvict() {
         ObjectStorage objectStorage = new MemoryObjectStorage();
-        LocalStreamRangeIndexCache cache = new LocalStreamRangeIndexCache();
+        LocalStreamRangeIndexCache cache = LocalStreamRangeIndexCache.create();
         cache.start();
         cache.init(NODE_0, objectStorage);
         int streamNum = 500;
@@ -180,7 +180,7 @@ public void testEvict() {
     @Test
     public void testCompact() {
         ObjectStorage objectStorage = new MemoryObjectStorage();
-        LocalStreamRangeIndexCache cache = new LocalStreamRangeIndexCache();
+        LocalStreamRangeIndexCache cache = LocalStreamRangeIndexCache.create();
         cache.start();
         cache.init(NODE_0, objectStorage);
         CommitStreamSetObjectRequest request = new CommitStreamSetObjectRequest();
@@ -260,7 +260,7 @@ public void testCompact() {
     @Test
     public void testCompactWithStreamDeleted() {
         ObjectStorage objectStorage = new MemoryObjectStorage();
-        LocalStreamRangeIndexCache cache = new LocalStreamRangeIndexCache();
+        LocalStreamRangeIndexCache cache = LocalStreamRangeIndexCache.create();
         cache.start();
         cache.init(NODE_0, objectStorage);
         CommitStreamSetObjectRequest request = new CommitStreamSetObjectRequest();