[#11350] Add write option to AsyncBufferedMutatorBuilder

collector/src/main/resources/hbase-root.properties

@@ -27,6 +27,11 @@ hbase.client.properties.hbase.client.retries.number=4
 hbase.client.put-writer.concurrency-limit=100000
 hbase.client.span-put-writer.concurrency-limit=0
 
+# asyncBufferedMutator, asyncTable
+hbase.client.put-writer=asyncBufferedMutator
+hbase.client.put-writer.async-buffered-mutator.writeBufferSize=100
+hbase.client.put-writer.async-buffered-mutator.writeBufferPeriodicFlush=100
+
 
 # hbase async =================================================================
 # enable hbase async operation. default: false

commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/async/AsyncBufferedMutatorCustomizer.java

@@ -0,0 +1,7 @@
+package com.navercorp.pinpoint.common.hbase.async;
+
+import org.apache.hadoop.hbase.client.AsyncBufferedMutatorBuilder;
+
+public interface AsyncBufferedMutatorCustomizer {
+    void customize(AsyncBufferedMutatorBuilder builder);
+}

commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/async/AsyncBufferedMutatorFactory.java

@@ -0,0 +1,13 @@
+package com.navercorp.pinpoint.common.hbase.async;
+
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.AsyncBufferedMutator;
+
+import java.util.concurrent.ExecutorService;
+
+public interface AsyncBufferedMutatorFactory {
+
+    AsyncBufferedMutator getBufferedMutator(TableName tableName, ExecutorService pool);
+
+    AsyncBufferedMutator getBufferedMutator(TableName tableName);
+}

commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/async/AsyncTableCustomizer.java

@@ -17,11 +17,8 @@
 
 package com.navercorp.pinpoint.common.hbase.async;
 
-import org.apache.hadoop.hbase.client.AsyncBufferedMutatorBuilder;
 import org.apache.hadoop.hbase.client.AsyncTableBuilder;
 
 public interface AsyncTableCustomizer {
     void customize(AsyncTableBuilder<?> builder);
-
-    void customize(AsyncBufferedMutatorBuilder builder);
 }

commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/async/AsyncTableFactory.java

@@ -19,7 +19,6 @@
 
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.AdvancedScanResultConsumer;
-import org.apache.hadoop.hbase.client.AsyncBufferedMutator;
 import org.apache.hadoop.hbase.client.AsyncTable;
 import org.apache.hadoop.hbase.client.ScanResultConsumer;
 
@@ -38,8 +37,4 @@ public interface AsyncTableFactory {
      */
     AsyncTable<ScanResultConsumer> getTable(TableName tableName, ExecutorService pool);
 
-
-    AsyncBufferedMutator getBufferedMutator(TableName tableName, ExecutorService pool);
-
-    AsyncBufferedMutator getBufferedMutator(TableName tableName);
 }

commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/async/BatchAsyncHbasePutWriter.java

@@ -26,9 +26,9 @@
 import java.util.concurrent.CompletableFuture;
 
 public class BatchAsyncHbasePutWriter implements HbasePutWriter {
-    private final AsyncTableFactory asyncTableFactory;
+    private final AsyncBufferedMutatorFactory asyncTableFactory;
 
-    public BatchAsyncHbasePutWriter(AsyncTableFactory asyncTableFactory) {
+    public BatchAsyncHbasePutWriter(AsyncBufferedMutatorFactory asyncTableFactory) {
         this.asyncTableFactory = Objects.requireNonNull(asyncTableFactory, "asyncTableFactory");
     }
 

commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/async/DefaultAsyncBufferedMutatorCustomizer.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2023 NAVER Corp.
+ *
+ * Licensed 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 com.navercorp.pinpoint.common.hbase.async;
+
+import org.apache.hadoop.hbase.client.AsyncBufferedMutatorBuilder;
+
+import java.util.concurrent.TimeUnit;
+
+public class DefaultAsyncBufferedMutatorCustomizer implements AsyncBufferedMutatorCustomizer {
+
+    private long writeBufferSize = 100;
+    private long writeBufferPeriodicFlush = 100;
+
+    public DefaultAsyncBufferedMutatorCustomizer() {
+    }
+
+    public void setWriteBufferSize(long writeBufferSize) {
+        this.writeBufferSize = writeBufferSize;
+    }
+
+    public void setWriteBufferPeriodicFlush(long writeBufferPeriodicFlush) {
+        this.writeBufferPeriodicFlush = writeBufferPeriodicFlush;
+    }
+
+    @Override
+    public void customize(AsyncBufferedMutatorBuilder builder) {
+        builder.setWriteBufferSize(writeBufferSize);
+        builder.setWriteBufferPeriodicFlush(writeBufferPeriodicFlush, TimeUnit.MILLISECONDS);
+    }
+
+    @Override
+    public String toString() {
+        return "DefaultAsyncBufferedMutatorCustomizer{" +
+                "writeBufferSize=" + writeBufferSize +
+                ", writeBufferPeriodicFlush=" + writeBufferPeriodicFlush +
+                '}';
+    }
+}

commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/async/DefaultAsyncTableCustomizer.java

@@ -17,21 +17,16 @@
 
 package com.navercorp.pinpoint.common.hbase.async;
 
-import org.apache.hadoop.hbase.client.AsyncBufferedMutatorBuilder;
 import org.apache.hadoop.hbase.client.AsyncTableBuilder;
 
-import java.util.concurrent.TimeUnit;
-
 public class DefaultAsyncTableCustomizer implements AsyncTableCustomizer {
 
-    @Override
-    public void customize(AsyncTableBuilder<?> builder) {
-
+    public DefaultAsyncTableCustomizer() {
     }
 
     @Override
-    public void customize(AsyncBufferedMutatorBuilder builder) {
+    public void customize(AsyncTableBuilder<?> builder) {
 
-        builder.setWriteBufferPeriodicFlush(500, TimeUnit.MILLISECONDS);
     }
+
 }

commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/async/HbaseAsyncBufferedMutatorFactory.java

@@ -0,0 +1,38 @@
+package com.navercorp.pinpoint.common.hbase.async;
+
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.AsyncBufferedMutator;
+import org.apache.hadoop.hbase.client.AsyncBufferedMutatorBuilder;
+import org.apache.hadoop.hbase.client.AsyncConnection;
+import org.springframework.cache.annotation.Cacheable;
+
+import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+
+public class HbaseAsyncBufferedMutatorFactory implements AsyncBufferedMutatorFactory {
+
+    private final AsyncConnection connection;
+    private final AsyncBufferedMutatorCustomizer customizer;
+
+    public HbaseAsyncBufferedMutatorFactory(AsyncConnection connection, AsyncBufferedMutatorCustomizer customizer) {
+        this.connection = Objects.requireNonNull(connection, "connection");
+        this.customizer = Objects.requireNonNull(customizer, "customizer");
+    }
+
+
+    @Override
+    @Cacheable(cacheNames = "bufferedMutator-pool", keyGenerator = "tableNameAndPoolKeyGenerator", cacheManager = "hbaseAsyncBufferedMutatorManager")
+    public AsyncBufferedMutator getBufferedMutator(TableName tableName, ExecutorService pool) {
+        AsyncBufferedMutatorBuilder builder = connection.getBufferedMutatorBuilder(tableName, pool);
+        customizer.customize(builder);
+        return builder.build();
+    }
+
+    @Override
+    @Cacheable(cacheNames = "bufferedMutator", keyGenerator = "tableNameAndPoolKeyGenerator", cacheManager = "hbaseAsyncBufferedMutatorManager")
+    public AsyncBufferedMutator getBufferedMutator(TableName tableName) {
+        AsyncBufferedMutatorBuilder builder = connection.getBufferedMutatorBuilder(tableName);
+        customizer.customize(builder);
+        return builder.build();
+    }
+}

commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/async/HbaseAsyncTableFactory.java

@@ -19,8 +19,6 @@
 
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.AdvancedScanResultConsumer;
-import org.apache.hadoop.hbase.client.AsyncBufferedMutator;
-import org.apache.hadoop.hbase.client.AsyncBufferedMutatorBuilder;
 import org.apache.hadoop.hbase.client.AsyncConnection;
 import org.apache.hadoop.hbase.client.AsyncTable;
 import org.apache.hadoop.hbase.client.AsyncTableBuilder;
@@ -56,21 +54,4 @@ public AsyncTable<ScanResultConsumer> getTable(TableName tableName, ExecutorServ
         return builder.build();
     }
 
-
-    @Override
-    @Cacheable(cacheNames = "bufferedMutator-pool", keyGenerator = "tableNameAndPoolKeyGenerator", cacheManager = "hbaseAsyncBufferedMutatorManager")
-    public AsyncBufferedMutator getBufferedMutator(TableName tableName, ExecutorService pool) {
-        AsyncBufferedMutatorBuilder builder = connection.getBufferedMutatorBuilder(tableName, pool);
-        customizer.customize(builder);
-        return builder.build();
-    }
-
-    @Override
-    @Cacheable(cacheNames = "bufferedMutator", keyGenerator = "tableNameAndPoolKeyGenerator", cacheManager = "hbaseAsyncBufferedMutatorManager")
-    public AsyncBufferedMutator getBufferedMutator(TableName tableName) {
-        AsyncBufferedMutatorBuilder builder = connection.getBufferedMutatorBuilder(tableName);
-        customizer.customize(builder);
-        return builder.build();
-    }
-
 }

commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/config/HbaseTemplateConfiguration.java

@@ -24,12 +24,16 @@
 import com.navercorp.pinpoint.common.hbase.HbaseTemplate;
 import com.navercorp.pinpoint.common.hbase.HbaseVersionCheckBean;
 import com.navercorp.pinpoint.common.hbase.TableFactory;
+import com.navercorp.pinpoint.common.hbase.async.AsyncBufferedMutatorCustomizer;
+import com.navercorp.pinpoint.common.hbase.async.AsyncBufferedMutatorFactory;
 import com.navercorp.pinpoint.common.hbase.async.AsyncHbasePutWriter;
 import com.navercorp.pinpoint.common.hbase.async.AsyncTableCustomizer;
 import com.navercorp.pinpoint.common.hbase.async.AsyncTableFactory;
 import com.navercorp.pinpoint.common.hbase.async.BatchAsyncHbasePutWriter;
 import com.navercorp.pinpoint.common.hbase.async.ConcurrencyDecorator;
+import com.navercorp.pinpoint.common.hbase.async.DefaultAsyncBufferedMutatorCustomizer;
 import com.navercorp.pinpoint.common.hbase.async.DefaultAsyncTableCustomizer;
+import com.navercorp.pinpoint.common.hbase.async.HbaseAsyncBufferedMutatorFactory;
 import com.navercorp.pinpoint.common.hbase.async.HbaseAsyncCacheConfiguration;
 import com.navercorp.pinpoint.common.hbase.async.HbaseAsyncTableFactory;
 import com.navercorp.pinpoint.common.hbase.async.HbasePutWriter;
@@ -78,6 +82,19 @@
         return new HbaseAsyncTableFactory(connection, customizer);
     }
 
+    @Bean
+    @ConfigurationProperties(prefix = "hbase.client.put-writer.async-buffered-mutator")
+    public AsyncBufferedMutatorCustomizer asyncBufferedMutatorCustomizer() {
+        return new DefaultAsyncBufferedMutatorCustomizer();
+    }
+
+    @Bean
+    public AsyncBufferedMutatorFactory hbaseAsyncBufferedMutatorFactory(@Qualifier("hbaseAsyncConnection") AsyncConnection connection,
+                                                                        AsyncBufferedMutatorCustomizer customizer) {
+        logger.info("AsyncBufferedMutatorCustomizer {}", customizer);
+        return new HbaseAsyncBufferedMutatorFactory(connection, customizer);
+    }
+
 
     @Bean
     @ConditionalOnProperty(name = "hbase.client.parallel.scan.enable", havingValue = "true")
@@ -136,7 +153,7 @@
         @Primary
         @Bean
         public HbasePutWriter hbasePutWriter(@Qualifier("hbaseAsyncTableFactory") AsyncTableFactory asyncTableFactory,
-                                                  @Qualifier("concurrencyDecorator") HbasePutWriterDecorator decorator) {
+                                             @Qualifier("concurrencyDecorator") HbasePutWriterDecorator decorator) {
             HbasePutWriter putWriter = newPutWriter(asyncTableFactory, decorator);
             logger.info("hbasePutWriter {}", putWriter);
             return putWriter;
@@ -149,7 +166,7 @@
 
         @Bean
         public HbasePutWriter spanPutWriter(@Qualifier("hbaseAsyncTableFactory") AsyncTableFactory asyncTableFactory,
-                                             @Qualifier("spanConcurrencyDecorator") HbasePutWriterDecorator decorator) {
+                                            @Qualifier("spanConcurrencyDecorator") HbasePutWriterDecorator decorator) {
             HbasePutWriter putWriter = newPutWriter(asyncTableFactory, decorator);
             logger.info("hbaseSpanPutWriter {}", putWriter);
             return putWriter;
@@ -179,8 +196,8 @@
 
         @Primary
         @Bean
-        public HbasePutWriter hbasePutWriter(@Qualifier("hbaseAsyncTableFactory") AsyncTableFactory asyncTableFactory,
-                                                  @Qualifier("concurrencyDecorator") HbasePutWriterDecorator decorator) {
+        public HbasePutWriter hbasePutWriter(@Qualifier("hbaseAsyncBufferedMutatorFactory") AsyncBufferedMutatorFactory asyncTableFactory,
+                                             @Qualifier("concurrencyDecorator") HbasePutWriterDecorator decorator) {
             HbasePutWriter hbasePutWriter = newPutWriter(asyncTableFactory, decorator);
             logger.info("hbasePutWriter {}", hbasePutWriter);
             return hbasePutWriter;
@@ -192,8 +209,8 @@
         }
 
         @Bean
-        public HbasePutWriter spanPutWriter(@Qualifier("hbaseAsyncTableFactory") AsyncTableFactory asyncTableFactory,
-                                             @Qualifier("spanConcurrencyDecorator") HbasePutWriterDecorator decorator) {
+        public HbasePutWriter spanPutWriter(@Qualifier("hbaseAsyncBufferedMutatorFactory") AsyncBufferedMutatorFactory asyncTableFactory,
+                                            @Qualifier("spanConcurrencyDecorator") HbasePutWriterDecorator decorator) {
             HbasePutWriter hbasePutWriter = newPutWriter(asyncTableFactory, decorator);
             logger.info("HbaseSpanPutWriter {}", hbasePutWriter);
             return hbasePutWriter;
@@ -204,7 +221,7 @@
             return new ConcurrencyDecorator(concurrency);
         }
 
-        private HbasePutWriter newPutWriter(AsyncTableFactory asyncTableFactory, HbasePutWriterDecorator decorator) {
+        private HbasePutWriter newPutWriter(AsyncBufferedMutatorFactory asyncTableFactory, HbasePutWriterDecorator decorator) {
             HbasePutWriter writer = new BatchAsyncHbasePutWriter(asyncTableFactory);
             HbasePutWriter putWriter = decorator.decorator(writer);
             return new LoggingHbasePutWriter(putWriter);