Fix the problem in send buffer and consumer.

Davidhua1996 · Davidhua1996 · commit 5f755e418723 · 2022-10-21T10:49:32.000+08:00
diff --git a/streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/StreamisRpcLogAppender.java b/streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/StreamisRpcLogAppender.java
@@ -18,6 +18,7 @@
 import org.apache.logging.log4j.core.layout.PatternLayout;
 
 import java.io.Serializable;
+import java.nio.Buffer;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
diff --git a/streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/config/StreamisLogAppenderConfigBuilder.java b/streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/config/StreamisLogAppenderConfigBuilder.java
@@ -17,8 +17,11 @@ public abstract class StreamisLogAppenderConfigBuilder {
      * @param senderConfig sender config
      * @return
      */
-    public abstract StreamisLogAppenderConfig build(String applicationName,
-                                             Filter filter, RpcLogSenderConfig senderConfig);
+    public StreamisLogAppenderConfig build(String applicationName,
+                                             Filter filter, RpcLogSenderConfig senderConfig){
+        return null;
+    }
 
-   abstract Map<String, String> loadConfigProps();
+
+   public abstract Map<String, String> loadConfigProps();
 }
diff --git a/streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/sender/AbstractRpcLogSender.java b/streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/sender/AbstractRpcLogSender.java
@@ -421,7 +421,7 @@ public int drainLogsTo(SendBuffer<T> sendBuffer, int maxElements) {
 
         @SuppressWarnings("unchecked")
         private int sendBuf(SendBuffer<T> sendBuffer, Object[] items, int takeIndex, int len){
-            int send = sendBuffer.writeBuf((T[]) items, takeIndex, len);
+            int send = sendBuffer.writeBuf(items, takeIndex, len);
             if (send < len){
                 // Buffer full exception
                 exceptionListener.onException(this, null, "The sender buffer is full," +
diff --git a/streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/sender/SendLogCacheConsumer.java b/streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/sender/SendLogCacheConsumer.java
@@ -53,13 +53,16 @@ public SendLogCacheConsumer(String id, SendLogCache<T> cache,
     public void run() {
         int remain;
         long expireTimeInMills = requireNewFlushTime();
+        int capacity = sendBuffer.capacity();
         while (!this.isTerminated) {
             try {
-                if ((expireTimeInMills > 0 && expireTimeInMills >= System.currentTimeMillis())
-                        || (remain = this.sendBuffer.remaining()) <= 0) {
+                remain = this.sendBuffer.remaining();
+                if ((expireTimeInMills > 0 && expireTimeInMills <= System.currentTimeMillis()) || remain <= 0) {
                     // Transient to the read mode
-                    sendBuffer.flip();
-                    onFlushAndSend(sendBuffer);
+                    if (remain < capacity) {
+                        sendBuffer.flip();
+                        onFlushAndSend(sendBuffer);
+                    }
                     expireTimeInMills = requireNewFlushTime();
                     if (sendBuffer.isReadMode()) {
                         // Clear the buffer and transient to the write mode, otherwise continue writing
@@ -84,11 +87,16 @@ public void run() {
                 if (this.isTerminated && e instanceof InterruptedException){
                     return;
                 } else {
-                   System.err.println("SendLogCacheConsumer[" + Thread.currentThread().getName() + "] occurred exception [" + e.getLocalizedMessage() + "]");
+                    System.err.println("SendLogCacheConsumer[" + Thread.currentThread().getName() + "] occurred exception [" + e.getLocalizedMessage() + "]");
                    // For the unknown exception clear the cache
                    sendBuffer.clear();
                    expireTimeInMills = requireNewFlushTime();
                 }
+                try {
+                    Thread.sleep(500);
+                } catch (InterruptedException ex) {
+                    // Ignore
+                }
             }
         }
     }
diff --git a/streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/sender/StreamisRpcLogSender.java b/streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/sender/StreamisRpcLogSender.java
@@ -30,8 +30,11 @@ public StreamisRpcLogSender(String applicationName, RpcLogSenderConfig rpcSender
     @Override
     protected StreamisLogEvents aggregateBuffer(SendBuffer<StreamisLogEvent> sendBuffer) {
         int remain = sendBuffer.remaining();
-        StreamisLogEvent[] logEvents = new StreamisLogEvent[remain];
-        sendBuffer.readBuf(logEvents, 0, logEvents.length);
-        return new StreamisLogEvents(applicationName, logEvents);
+        if (remain > 0) {
+            StreamisLogEvent[] logEvents = new StreamisLogEvent[remain];
+            sendBuffer.readBuf(logEvents, 0, logEvents.length);
+            return new StreamisLogEvents(applicationName, logEvents);
+        }
+        return null;
     }
 }
diff --git a/streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/sender/buf/AbstractSendBuffer.java b/streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/sender/buf/AbstractSendBuffer.java
@@ -103,6 +103,8 @@ final int nextPosition(int offset, Flag accessFlag){
         }
         if (p + offset > limit){
             this.position = limit;
+        } else {
+            this.position = p + offset;
         }
         return p;
     }
diff --git a/streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/sender/buf/ImmutableSendBuffer.java b/streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/sender/buf/ImmutableSendBuffer.java
@@ -31,7 +31,7 @@ public void capacity(String newCapacity) {
 
     @Override
     @SuppressWarnings("all")
-    public int writeBuf(E[] elements, int srcIndex, int length) {
+    public int writeBuf(Object[] elements, int srcIndex, int length) {
         if (srcIndex < elements.length){
             int startPos = nextPosition(Math.min(elements.length - srcIndex, length), Flag.WRITE_MODE);
             if (startPos >= 0){
@@ -45,7 +45,7 @@ public int writeBuf(E[] elements, int srcIndex, int length) {
 
     @Override
     @SuppressWarnings("all")
-    public int readBuf(E[] elements, int srcIndex, int length) {
+    public int readBuf(Object[] elements, int srcIndex, int length) {
         if (srcIndex < elements.length){
             int startPos = nextPosition(Math.min(elements.length - srcIndex, length), Flag.READ_MODE);
             if (startPos >= 0){
diff --git a/streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/sender/buf/SendBuffer.java b/streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/sender/buf/SendBuffer.java
@@ -64,7 +64,7 @@ public interface SendBuffer<E> {
      * @param length the length to read
      * @return write num
      */
-    int writeBuf(E[] elements, int srcIndex, int length);
+    int writeBuf(Object[] elements, int srcIndex, int length);
 
     /**
      * Read buffer element
@@ -79,7 +79,7 @@ public interface SendBuffer<E> {
      * @param length the length to write
      * @return read num
      */
-    int readBuf(E[] elements, int srcIndex, int length);
+    int readBuf(Object[] elements, int srcIndex, int length);
 
     /**
      * Compact the buffer, avoid the useless elements
diff --git a/streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/sender/http/AbstractHttpLogSender.java b/streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/sender/http/AbstractHttpLogSender.java
@@ -15,6 +15,7 @@
 import java.io.IOException;
 import java.io.InterruptedIOException;
 import java.net.UnknownHostException;
+import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
@@ -55,14 +56,15 @@ public int retryCount() {
 
             @Override
             public SendLogExceptionStrategy.RetryDescription onException(Exception e, SendBuffer<T> sendBuffer) {
+                e.printStackTrace();
                 boolean shouldRetry = false;
                 // Limit of exception number is the same as the retry times
                 if (exceptionCounter.incrementAndGet() > retryCount()){
                     serverRecoveryTimePoint.set(System.currentTimeMillis() +
                             TimeUnit.SECONDS.toMillis(rpcSenderConfig.getServerRecoveryTimeInSec()));
                 } else {
                     for (Class<?> retryOnException : retryOnExceptions) {
-                        if (retryOnException.isAssignableFrom(e.getClass())) {
+                        if (retryOnException.equals(e.getClass())) {
                             shouldRetry = true;
                             break;
                         }
@@ -76,7 +78,7 @@ public SendLogExceptionStrategy.RetryDescription onException(Exception e, SendBu
                     sendBuffer.compact( element -> element.mark() > 1);
                     shouldRetry = false;
                 }
-                exceptionListener.onException(sender, e, null);
+                Optional.ofNullable(exceptionListener).ifPresent(listener -> listener.onException(sender, e, null));
                 return new RetryDescription(shouldRetry);
             }
         };
@@ -97,6 +99,7 @@ protected void doSend(E aggregatedEntity, RpcLogSenderConfig rpcSenderConfig) th
                     return;
                 }
             }
+
             EntityPostAction<E> postAction = new EntityPostAction<>(rpcSenderConfig.getAddress(), aggregatedEntity);
             RpcAuthConfig authConfig = rpcSenderConfig.getAuthConfig();
             postAction.getRequestHeaders().put(authConfig.getTokenUserKey(), authConfig.getTokenUser());
diff --git a/streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/test/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/StreamisLogAppenderTest.java b/streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/test/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/StreamisLogAppenderTest.java
@@ -0,0 +1,16 @@
+package com.webank.wedatasphere.streamis.jobmanager.log.collector;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class StreamisLogAppenderTest {
+    private static final Logger LOG = LoggerFactory.getLogger(StreamisLogAppenderTest.class);
+    public static void main(String[] args) throws InterruptedException {
+        while(true){
+            for(int i = 0; i < 100; i ++){
+                LOG.info("Stream Log appender test");
+            }
+            Thread.sleep(1000);
+        }
+    }
+}
diff --git a/streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/test/resources/log4j2.xml b/streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/test/resources/log4j2.xml
@@ -0,0 +1,37 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Copyright 2019 WeBank
+  ~
+  ~ 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.
+  ~
+  -->
+
+<configuration status="error" monitorInterval="30">
+    <appenders>
+        <StreamRpcLog name="StreamRpcLog" appName="stream_application">
+            <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n"/>
+            <RpcLogSender address="http://127.0.0.1/log/collector" sendRetryCnt="3"
+                          connectionTimeout="3000" socketTimeout="15000" serverRecoveryTimeInSec="5" maxDelayTimeInSec="60">
+                <AuthConfig tokenCodeKey="" tokenCode="" tokenUser="" tokenUserKey=""/>
+                <SendLogCache size="200" maxConsumeThread="10"/>
+                <SendBuffer size="50" expireTimeInSec="2"/>
+            </RpcLogSender>
+        </StreamRpcLog>`
+    </appenders>
+    <loggers>
+        <root level="INFO" additivity="true">
+<!--            <appender-ref ref="Console"/>-->
+            <appender-ref ref="StreamRpcLog"/>
+        </root>
+    </loggers>
+</configuration>

Original file line number	Diff line number	Diff line change
`@@ -103,6 +103,8 @@ final int nextPosition(int offset, Flag accessFlag){`
`103`	`103`	`}`
`104`	`104`	`if (p + offset > limit){`
`105`	`105`	`this.position = limit;`
	`106`	`+ } else {`
	`107`	`+ this.position = p + offset;`
`106`	`108`	`}`
`107`	`109`	`return p;`
`108`	`110`	`}`