Merge pull request #1 from volcengine/producer_close

Producer close

Author: StudyForZX

.gitignore

@@ -8,4 +8,8 @@ scripts/build.out
 *.swp
 .classpath
 .factorypath
-.project
+.project
+
+# vscode
+.vscode
+.devcontainer

README.md

@@ -1,5 +1,5 @@
 <h1 align="center"><img src="https://iam.volccdn.com/obj/volcengine-public/pic/volcengine-icon.png"></h1>
-<h1 align="center">火山引擎Android SDK for TLS</h1> 
+<h1 align="center">火山引擎Android SDK for TLS</h1>
 欢迎使用火山引擎SDK for Android，本文档为您介绍如何获取及调用SDK。
 
 ## 前置准备
@@ -28,7 +28,7 @@ Key。更多信息可参考[访问密钥帮助文档](https://www.volcengine.com
 1. 创建安卓项目。
 2. Gradle配置mavenCentral()，并引入SDK。
 ```xml
-   implementation 'com.volcengine:volc-tls-android-sdk:1.1.4'
+   implementation 'com.volcengine:volc-tls-android-sdk:1.1.5'
 ```
 如果有依赖冲突，请使用指定你需要的版本（以okhttp为例子）
 ```xml

pom.xml

@@ -4,7 +4,7 @@
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.volcengine</groupId>
-    <version>1.1.4</version>
+    <version>1.1.5</version>
     <artifactId>volc-tls-android-sdk</artifactId>
     <name>volc-tls-android-sdk</name>
     <url>https://www.volcengine.com/docs/6470/516236</url>
@@ -207,11 +207,11 @@
     <distributionManagement>
         <snapshotRepository>
             <id>sonatype-nexus-staging</id>
-            <url>https://oss.sonatype.org/content/repositories/snapshots</url>
+            <url>https://ossrh-staging-api.central.sonatype.com/content/repositories/snapshots</url>
         </snapshotRepository>
         <repository>
             <id>sonatype-nexus-staging</id>
-            <url>https://oss.sonatype.org/service/local/staging/deploy/maven2/</url>
+            <url>https://ossrh-staging-api.central.sonatype.com/service/local/staging/deploy/maven2/</url>
         </repository>
     </distributionManagement>
 </project>

src/main/java/com/volcengine/model/tls/producer/BatchLog.java

@@ -30,6 +30,12 @@ public class BatchLog implements Delayed {
     EvictingQueue<Attempt> reservedAttempts;
     int attemptCount;
     long createMs;
+    long nextRetryMs;
+    long retryBackoffMs;
+    long maxRetryBackoffMs;
+    long baseRetryBackoffMs;
+    long baseIncreaseBackoffMs;
+
     private static final Logger LOG = LoggerFactory.getLogger(BatchLog.class);
 
     private BatchLog() {
@@ -43,6 +49,10 @@ public BatchLog(BatchKey batchKey, ProducerConfig producerConfig) {
         this.attemptCount = 0;
         this.reservedAttempts = EvictingQueue.create(producerConfig.getMaxReservedAttempts());
         this.createMs = System.currentTimeMillis();
+        this.retryBackoffMs = 0;
+        this.maxRetryBackoffMs = 10 * 1000;
+        this.baseRetryBackoffMs = 1000;
+        this.baseIncreaseBackoffMs = 1000;
     }
 
     public boolean tryAdd(PutLogRequest.LogGroup logGroup, int batchSize, CallBack callBack) {
@@ -89,6 +99,17 @@ public synchronized void fireCallbacks() {
         fireCallbacks(result);
     }
 
+    public void handleNextTry() {
+        if (attemptCount == 1) {
+            retryBackoffMs += baseRetryBackoffMs;
+        } else {
+            double increaseBackoffMs = Math.random() * baseIncreaseBackoffMs;
+            retryBackoffMs += (long) increaseBackoffMs;
+        }
+        retryBackoffMs = Math.min(retryBackoffMs, maxRetryBackoffMs);
+        nextRetryMs = System.currentTimeMillis() + retryBackoffMs;
+    }
+
     private void fireCallbacks(Result result) {
         for (CallBack callBack : callBackList) {
             callBack.onComplete(result);
@@ -97,12 +118,12 @@ private void fireCallbacks(Result result) {
 
     @Override
     public int compareTo(Delayed o) {
-        return 0;
+        return (int) (nextRetryMs - ((BatchLog) o).getNextRetryMs());
     }
 
     @Override
     public long getDelay(TimeUnit unit) {
-        return unit.convert(producerConfig.getLingerMs(), TimeUnit.MILLISECONDS);
+        return unit.convert(nextRetryMs - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
     }
 
     @Data
@@ -169,11 +190,14 @@ public String toString() {
                 "batchKey=" + batchKey +
                 ", currentBatchSize=" + currentBatchSize +
                 ", currentBatchCount=" + currentBatchCount +
-                ", callBackList=" + callBackList +
-                ", producerConfig=" + producerConfig +
                 ", reservedAttempts=" + reservedAttempts +
                 ", attemptCount=" + attemptCount +
                 ", createMs=" + createMs +
+                ", nextRetryMs=" + nextRetryMs +
+                ", retryBackoffMs=" + retryBackoffMs +
+                ", maxRetryBackoffMs=" + nextRetryMs +
+                ", baseRetryBackoffMs=" + baseRetryBackoffMs +
+                ", baseIncreaseBackoffMs=" + baseIncreaseBackoffMs +
                 '}';
     }
 }

src/main/java/com/volcengine/service/tls/BatchHandler.java

@@ -4,6 +4,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -23,12 +25,17 @@ public BatchHandler(String name, Semaphore memoryLock, BlockingQueue<BatchLog> b
         this.batchQueue = batchQueue;
         this.batchCount = batchCount;
         this.name = name;
-        this.closed = false;
     }
 
     @Override
     public void run() {
         handleBatches();
+
+        List<BatchLog> batchLogList = new ArrayList<>();
+        batchQueue.drainTo(batchLogList);
+        for (BatchLog batchLog : batchLogList) {
+            handle(batchLog);
+        }
     }
 
     private void handleBatches() {
@@ -43,9 +50,20 @@ private void handleBatches() {
     }
 
     private void handle(BatchLog batch) {
-        batch.fireCallbacks();
-        batchCount.decrementAndGet();
-        memoryLock.release(batch.getCurrentBatchSize());
+        try {
+            batch.fireCallbacks();
+        } catch (Throwable t) {
+            LOG.error("batch fire callbacks failed: ", t);
+        } finally {
+            batchCount.decrementAndGet();
+            memoryLock.release(batch.getCurrentBatchSize());
+        }
+    }
+
+    @Override
+    public void start() {
+        this.closed = false;
+        super.start();
     }
 
     public void close() {

src/main/java/com/volcengine/service/tls/LogDispatcher.java

@@ -25,8 +25,7 @@ public class LogDispatcher {
     private final BlockingQueue<BatchLog> successQueue;
     private final BlockingQueue<BatchLog> failureQueue;
     private static final Logger LOG = LoggerFactory.getLogger(LogDispatcher.class);
-    private final AtomicInteger addLogLock = new AtomicInteger(0);
-    private volatile boolean closed = true;
+    private volatile boolean closed;
     private final Semaphore memoryLock;
     private final AtomicInteger batchCount;
     private final RetryManager retryManager;
@@ -50,7 +49,6 @@ public LogDispatcher(ProducerConfig producerConfig, String producerName, Blockin
         this.batchCount = batchCount;
         this.retryManager = retryManager;
         this.client = ClientBuilder.newClient(producerConfig.getClientConfig());
-        this.closed = false;
     }
 
     public TLSLogClient getClient() {
@@ -62,19 +60,19 @@ public ConcurrentHashMap<BatchLog.BatchKey, BatchLog.BatchManager> getBatches()
     }
 
     public void start() {
+        this.closed = false;
         LOG.info(String.format("log dispatcher %s started and client init success", producerName));
     }
 
     public ExecutorService getExecutorService() {
         return this.executorService;
     }
 
-    public void close() throws InterruptedException, LogException {
-        executorService.shutdown();
+    public void close() {
         this.closed = true;
     }
 
-    public void closeNow() throws InterruptedException, LogException {
+    public void closeNow() {
         executorService.shutdownNow();
         this.closed = true;
     }
@@ -99,9 +97,7 @@ public void resetAccessKeyToken(String accessKey, String secretKey, String secur
 
     public void addBatch(String hashKey, String topicId, String source, String filename,
                          PutLogRequest.LogGroup logGroup, CallBack callBack) throws InterruptedException, LogException {
-        addLogLock.incrementAndGet();
         doAdd(hashKey, topicId, source, filename, logGroup, callBack);
-        addLogLock.decrementAndGet();
     }
 
     private void doAdd(String hashKey, String topicId, String source, String filename,
@@ -121,6 +117,8 @@ private void doAdd(String hashKey, String topicId, String source, String filenam
         } else {
             boolean acquired = memoryLock.tryAcquire(batchSize, maxBlockMs, TimeUnit.MILLISECONDS);
             if (!acquired) {
+                LOG.warn(String.format("Failed to acquire memory within the configured max blocking time %d ms, requiredSizeInBytes=%d, availableSizeInBytes=%d",
+                        producerConfig.getMaxBlockMs(), batchSize, memoryLock.availablePermits()));
                 throw new LogException("Producer Error", "dispatcher %s try acquire memory lock failed", null);
             }
         }
@@ -132,9 +130,8 @@ private void doAdd(String hashKey, String topicId, String source, String filenam
                 addToBatchManager(batchKey, logGroup, callBack, batchSize, batchManager);
             }
         } catch (Exception e) {
-            throw new LogException("Producer Error", "dispatcher add batch concurrent error", null);
-        } finally {
             memoryLock.release(batchSize);
+            throw new LogException("Producer Error", "dispatcher add batch concurrent error", null);
         }
     }
 
@@ -146,7 +143,7 @@ private int calculateSize(PutLogRequest.LogGroup logGroup) {
     }
 
     private void addToBatchManager(BatchLog.BatchKey batchKey, PutLogRequest.LogGroup logGroup, CallBack callBack,
-                                   int batchSize, BatchLog.BatchManager batchManager) {
+                                   int batchSize, BatchLog.BatchManager batchManager) throws LogException {
         // add to exist batch
         BatchLog batchLog = batchManager.getBatchLog();
         if (batchLog != null) {
@@ -164,7 +161,12 @@ private void addToBatchManager(BatchLog.BatchKey batchKey, PutLogRequest.LogGrou
         // no batch create new and try send
         batchLog = new BatchLog(batchKey, producerConfig);
         batchManager.setBatchLog(batchLog);
-        batchLog.tryAdd(logGroup, batchSize, callBack);
+        boolean success = batchLog.tryAdd(logGroup, batchSize, callBack);
+        if (!success) {
+            LOG.error(String.format("tryAdd batchLog failed, batchKey = %s, batchSize = %d, batchCount = %d",
+                    batchKey.toString(), batchSize, logGroup.getLogsCount()));
+            throw new LogException("Producer Error", "tryAdd batchLog failed", null);
+        }
         if (batchManager.fullAndSendBatchRequest()) {
             batchManager.addNow(producerConfig, executorService, client, successQueue, failureQueue, batchCount,
                     retryManager);

src/main/java/com/volcengine/service/tls/Mover.java

@@ -40,6 +40,12 @@ public Mover(String name, ProducerConfig producerConfig, LogDispatcher dispatche
     @Override
     public void run() {
         handlerTimeout();
+
+        handleRemainingBatch();
+        List<BatchLog> remainingRetryBatches = retryManager.handleRemainingBatches();
+        for (BatchLog log : remainingRetryBatches) {
+            executorService.submit(new SendBatchTask(log, producerConfig, successQueue, failureQueue, client, retryManager));
+        }
     }
 
     private void handlerTimeout() {
@@ -84,6 +90,24 @@ private long handleTimeoutBatch() {
         return remains;
     }
 
+    private void handleRemainingBatch() {
+        LOG.debug("mover" + name + "handler remaining batch");
+        List<BatchLog> batchLogs = new ArrayList<>();
+        for (Map.Entry<BatchLog.BatchKey, BatchLog.BatchManager> entry : batches.entrySet()) {
+            BatchLog.BatchManager batchManager = entry.getValue();
+            synchronized (batchManager) {
+                BatchLog batchLog = batchManager.getBatchLog();
+                if (batchLog == null) {
+                    continue;
+                }
+                batchManager.removeBatch(batchLogs);
+            }
+        }
+        for (BatchLog log : batchLogs) {
+            executorService.submit(new SendBatchTask(log, producerConfig, successQueue, failureQueue, client, retryManager));
+        }
+    }
+
     public void close() {
         this.closed = true;
         super.interrupt();

src/main/java/com/volcengine/service/tls/Producer.java

@@ -73,6 +73,15 @@ void sendLogsV2(String hashKey, String topicId, String source, String filename,
      */
     void close() throws InterruptedException, LogException;
 
+    /**
+     * 暂停Producer
+     *
+     * @param timeoutMs 超时等待时间
+     * @throws InterruptedException
+     * @throws LogException
+     */
+    void close(long timeoutMs) throws InterruptedException, LogException;
+
     /**
      * 立刻暂停producer
      *
@@ -88,4 +97,4 @@ void sendLogsV2(String hashKey, String topicId, String source, String filename,
      * @throws LogException
      */
     void config(ProducerConfig producerConfig) throws LogException;
-}
+}