[#10976] Add additional spanPutWriter

collector/src/main/java/com/navercorp/pinpoint/collector/config/HbaseAsyncConfiguration.java

@@ -36,20 +36,19 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
+import org.springframework.context.annotation.Primary;
 
 import java.util.List;
 
 @Configuration
-@Import(CommonCacheManagerConfiguration.class)
+@Import({
+        CommonCacheManagerConfiguration.class,
+        HbaseAsyncConfiguration.TableMultiplexerConfig.class
+})
 public class HbaseAsyncConfiguration {
 
     private final Logger logger = LogManager.getLogger(HbaseAsyncConfiguration.class);
 
-    @Bean
-    public HbaseMultiplexerProperties hbaseMultiplexerProperties() {
-        return new HbaseMultiplexerProperties();
-    }
-
     @Bean
     public HbaseBatchPerformanceCounter batchPerformanceCounter() {
         return new HbaseBatchPerformanceCounter();
@@ -60,16 +59,43 @@
         return new HBaseAsyncOperationMetrics(hBaseAsyncOperationList);
     }
 
-    @Bean
+    @Configuration
     @ConditionalOnProperty(name = "hbase.client.put-writer", havingValue = "tableMultiplexer")
-    public HbasePutWriter tableMultiplexerBatchWriter(@Qualifier("batchConnectionFactory") Connection connection,
-                                                           HbaseBatchPerformanceCounter counter,
-                                                           HbaseMultiplexerProperties properties) throws Exception {
-        HBaseTableMultiplexerFactory factory = new HBaseTableMultiplexerFactory(connection, counter);
-        factory.setHbaseMultiplexerProperties(properties);
-        HbasePutWriter writer = factory.getObject();
-        logger.info("HbasePutWriter:{}", writer);
-        return new LoggingHbasePutWriter(writer);
+    public static class TableMultiplexerConfig {
+        private final Logger logger = LogManager.getLogger(TableMultiplexerConfig.class);
+
+        @Bean
+        public HbaseMultiplexerProperties hbaseMultiplexerProperties() {
+            return new HbaseMultiplexerProperties();
+        }
+
+        @Primary
+        @Bean
+        public HbasePutWriter hbasePutWriter(@Qualifier("batchConnectionFactory") Connection connection,
+                                             HbaseBatchPerformanceCounter counter,
+                                             HbaseMultiplexerProperties properties) throws Exception {
+            HbasePutWriter writer = newPutWriter(connection, counter, properties);
+            logger.info("HbaseSpanPutWriter:{}", writer);
+            return writer;
+        }
+
+        @Bean
+        public HbasePutWriter spanPutWriter(@Qualifier("batchConnectionFactory") Connection connection,
+                                             HbaseBatchPerformanceCounter counter,
+                                             HbaseMultiplexerProperties properties) throws Exception {
+            HbasePutWriter writer = newPutWriter(connection, counter, properties);
+            logger.info("HbaseSpanPutWriter:{}", writer);
+            return writer;
+        }
+
+        private HbasePutWriter newPutWriter(Connection connection, HbaseBatchPerformanceCounter counter,
+                                            HbaseMultiplexerProperties properties) throws Exception {
+            HBaseTableMultiplexerFactory factory = new HBaseTableMultiplexerFactory(connection, counter);
+            factory.setHbaseMultiplexerProperties(properties);
+            HbasePutWriter writer = factory.getObject();
+            return new LoggingHbasePutWriter(writer);
+        }
+
     }
 
 

collector/src/main/java/com/navercorp/pinpoint/collector/dao/hbase/HbaseTraceDaoV2.java

@@ -62,7 +62,8 @@ public class HbaseTraceDaoV2 implements TraceDao {
     private final RowKeyEncoder<TransactionId> rowKeyEncoder;
     private final HbasePutWriter putWriter;
 
-    public HbaseTraceDaoV2(HbasePutWriter putWriter,
+    public HbaseTraceDaoV2(@Qualifier("spanPutWriter")
+                           HbasePutWriter putWriter,
                            TableNameProvider tableNameProvider,
                            @Qualifier("traceRowKeyEncoderV2") RowKeyEncoder<TransactionId> rowKeyEncoder,
                            SpanSerializerV2 spanSerializer,

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

@@ -23,8 +23,9 @@ hbase.client.properties.hbase.client.max.retries.in.queue=10000
 hbase.client.properties.hbase.client.async.max.retries.in.queue=10000
 hbase.client.properties.hbase.client.retries.number=4
 
-# Limit concurrent requests to prevent OOM. default: 100000
+# Limit concurrent requests to prevent OOM. Unlimited:0
 hbase.client.put-writer.concurrency-limit=100000
+hbase.client.span-put-writer.concurrency-limit=0
 
 # hbase async =================================================================
 # enable hbase async operation. default: false

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

@@ -55,4 +55,10 @@
     }
 
 
+    @Override
+    public String toString() {
+        return "LoggingHbasePutWriter{" +
+                "delegate=" + delegate +
+                '}';
+    }
 }

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

@@ -54,7 +54,12 @@
 import java.util.Optional;
 
 @org.springframework.context.annotation.Configuration
-@Import(HbaseAsyncCacheConfiguration.class)
+@Import({
+        HbaseAsyncCacheConfiguration.class,
+
+        HbaseTemplateConfiguration.AsyncBufferedHbasePutWriterConfig.class,
+        HbaseTemplateConfiguration.AsyncHbasePutWriterConfig.class,
+})
 public class HbaseTemplateConfiguration {
     private final Logger logger = LogManager.getLogger(HbaseTemplateConfiguration.class);
 
@@ -118,30 +123,91 @@
         return template2;
     }
 
-    @Bean
+    @org.springframework.context.annotation.Configuration
     @ConditionalOnProperty(name = "hbase.client.put-writer", havingValue = "asyncTable")
-    public HbasePutWriter asyncHbasePutWriter(@Qualifier("hbaseAsyncTableFactory") AsyncTableFactory asyncTableFactory,
-                                              @Qualifier("concurrencyDecorator") HbasePutWriterDecorator decorator) {
-        AsyncHbasePutWriter writer = new AsyncHbasePutWriter(asyncTableFactory);
-        HbasePutWriter putWriter = decorator.decorator(writer);
-        logger.info("HbasePutWriter {}", putWriter);
-        return new LoggingHbasePutWriter(putWriter);
-    }
+    public static class AsyncHbasePutWriterConfig {
+        private final Logger logger = LogManager.getLogger(AsyncHbasePutWriterConfig.class);
 
-    @Bean
-    public HbasePutWriterDecorator concurrencyDecorator(@Value("${hbase.client.put-writer.concurrency-limit:1000000}") int concurrency) {
-        return new ConcurrencyDecorator(concurrency);
+        public AsyncHbasePutWriterConfig() {
+            logger.info("Install {}", AsyncHbasePutWriterConfig.class.getSimpleName());
+        }
+
+        @Primary
+        @Bean
+        public HbasePutWriter hbasePutWriter(@Qualifier("hbaseAsyncTableFactory") AsyncTableFactory asyncTableFactory,
+                                                  @Qualifier("concurrencyDecorator") HbasePutWriterDecorator decorator) {
+            HbasePutWriter putWriter = newPutWriter(asyncTableFactory, decorator);
+            logger.info("hbasePutWriter {}", putWriter);
+            return putWriter;
+        }
+
+        @Bean
+        public HbasePutWriterDecorator concurrencyDecorator(@Value("${hbase.client.put-writer.concurrency-limit:100000}") int concurrency) {
+            return new ConcurrencyDecorator(concurrency);
+        }
+
+        @Bean
+        public HbasePutWriter spanPutWriter(@Qualifier("hbaseAsyncTableFactory") AsyncTableFactory asyncTableFactory,
+                                             @Qualifier("spanConcurrencyDecorator") HbasePutWriterDecorator decorator) {
+            HbasePutWriter putWriter = newPutWriter(asyncTableFactory, decorator);
+            logger.info("hbaseSpanPutWriter {}", putWriter);
+            return putWriter;
+        }
+
+        @Bean
+        public HbasePutWriterDecorator spanConcurrencyDecorator(@Value("${hbase.client.span-put-writer.concurrency-limit:1000000}") int concurrency) {
+            return new ConcurrencyDecorator(concurrency);
+        }
+
+        private HbasePutWriter newPutWriter(AsyncTableFactory asyncTableFactory, HbasePutWriterDecorator decorator) {
+            HbasePutWriter writer = new AsyncHbasePutWriter(asyncTableFactory);
+            HbasePutWriter putWriter = decorator.decorator(writer);
+            return new LoggingHbasePutWriter(putWriter);
+        }
     }
 
 
-    @Bean
+    @org.springframework.context.annotation.Configuration
     @ConditionalOnProperty(name = "hbase.client.put-writer", havingValue = "asyncBufferedMutator", matchIfMissing = true)
-    public HbasePutWriter batchHbasePutWriter(@Qualifier("hbaseAsyncTableFactory") AsyncTableFactory asyncTableFactory,
-                                              @Qualifier("concurrencyDecorator") HbasePutWriterDecorator decorator) {
-        BatchAsyncHbasePutWriter writer = new BatchAsyncHbasePutWriter(asyncTableFactory);
-        HbasePutWriter putWriter = decorator.decorator(writer);
-        logger.info("HbasePutWriter {}", putWriter);
-        return new LoggingHbasePutWriter(putWriter);
+    public static class AsyncBufferedHbasePutWriterConfig {
+        private final Logger logger = LogManager.getLogger(AsyncBufferedHbasePutWriterConfig.class);
+
+        public AsyncBufferedHbasePutWriterConfig() {
+            logger.info("Install {}", AsyncBufferedHbasePutWriterConfig.class.getSimpleName());
+        }
+
+        @Primary
+        @Bean
+        public HbasePutWriter hbasePutWriter(@Qualifier("hbaseAsyncTableFactory") AsyncTableFactory asyncTableFactory,
+                                                  @Qualifier("concurrencyDecorator") HbasePutWriterDecorator decorator) {
+            HbasePutWriter hbasePutWriter = newPutWriter(asyncTableFactory, decorator);
+            logger.info("hbasePutWriter {}", hbasePutWriter);
+            return hbasePutWriter;
+        }
+
+        @Bean
+        public HbasePutWriterDecorator concurrencyDecorator(@Value("${hbase.client.put-writer.concurrency-limit:100000}") int concurrency) {
+            return new ConcurrencyDecorator(concurrency);
+        }
+
+        @Bean
+        public HbasePutWriter spanPutWriter(@Qualifier("hbaseAsyncTableFactory") AsyncTableFactory asyncTableFactory,
+                                             @Qualifier("spanConcurrencyDecorator") HbasePutWriterDecorator decorator) {
+            HbasePutWriter hbasePutWriter = newPutWriter(asyncTableFactory, decorator);
+            logger.info("HbaseSpanPutWriter {}", hbasePutWriter);
+            return hbasePutWriter;
+        }
+
+        @Bean
+        public HbasePutWriterDecorator spanConcurrencyDecorator(@Value("${hbase.client.span-put-writer.concurrency-limit:1000000}") int concurrency) {
+            return new ConcurrencyDecorator(concurrency);
+        }
+
+        private HbasePutWriter newPutWriter(AsyncTableFactory asyncTableFactory, HbasePutWriterDecorator decorator) {
+            HbasePutWriter writer = new BatchAsyncHbasePutWriter(asyncTableFactory);
+            HbasePutWriter putWriter = decorator.decorator(writer);
+            return new LoggingHbasePutWriter(putWriter);
+        }
     }
 
     @Bean