aliyun / aliyun-log-java-producer Goto Github PK

public boolean isMeetSendCondition() {
return curBatchSizeInBytes >= batchSizeThresholdInBytes || curBatchCount >= batchCountThreshold;
}
判断的意思是现在发送数据大小大于设定 或者 现在batch count 大于设定 为true
那这样的话很难为true啊 都进入阻塞队列了啊？

我们在接入aliyunlog时开发了一个快速接入的spring boot starter组件，欢迎体验
项目地址：https://github.com/kekingcn/aliyunlog-spring-boot-starter

Flink环境说明:

阿里云:
作业类型: FLINK_STREAM/DATASTREAM
Flink版本: blink-3.4.4

Maven配置如下:

          <dependency>
            <groupId>com.aliyun.openservices</groupId>
            <artifactId>flink-log-connector</artifactId>
            <version>0.1.24</version>
            <exclusions>
                <exclusion>
                    <artifactId>guava</artifactId>
                    <groupId>com.google.guava</groupId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>com.google.protobuf</groupId>
            <artifactId>protobuf-java</artifactId>
            <version>2.5.0</version>
        </dependency>
        <dependency>
            <groupId>com.google.guava</groupId>
            <artifactId>guava</artifactId>
            <version>28.0-jre</version>
        </dependency>

异常如下:

2021-02-09 16:01:44
java.lang.NoSuchMethodError: com.aliyun.openservices.aliyun.log.producer.Producer.send(Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;Ljava/util/List;)Lcom/google/common/util/concurrent/ListenableFuture;
	at com.aliyun.openservices.log.flink.FlinkLogProducer.invoke(FlinkLogProducer.java:158)
	at org.apache.flink.streaming.api.operators.StreamSink.processElement(StreamSink.java:56)
	at org.apache.flink.streaming.runtime.io.StreamInputProcessor.lambda$processRecordOrMark$1(StreamInputProcessor.java:293)
	at org.apache.flink.util.checkpointlock.SynchronizedLockHelper.lockAndRun(SynchronizedLockHelper.java:51)
	at org.apache.flink.util.checkpointlock.CheckpointLockDelegate.lockAndRun(CheckpointLockDelegate.java:52)
	at org.apache.flink.streaming.runtime.io.StreamInputProcessor.processRecordOrMark(StreamInputProcessor.java:290)
	at org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(StreamInputProcessor.java:237)
	at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:113)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:398)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:736)
	at java.lang.Thread.run(Thread.java:834)

建议把LogProducer.java 里面的 private static final Logger LOGGER = LoggerFactory.getLogger(LogProducer.class);放在static 变量最后，防止LOGGER 初始化时依赖LogProducer自己导致INSTANCE_ID_GENERATOR没初始化报错。
我在测试的时候发现，如果直接实例化LogProducer 来使用的话，log4j2可能还没有初始化，这时如果log4j2 配置了LogProducer 作为appender就会报INSTANCE_ID_GENERATOR NPE。

"aliyun-log-producer-234-timeout-thread-0" #1240 daemon prio=5 os_prio=0 tid=0x00007f6770002000 nid=0x73fee waiting on condition [0x00007f6728016000]
"aliyun-log-producer-234-io-thread-0" #1239 daemon prio=5 os_prio=0 tid=0x00007f69c8b3b800 nid=0x73fed waiting on condition [0x00007f6728117000]
"aliyun-log-producer-234-failure-batch-handler" #1238 daemon prio=5 os_prio=0 tid=0x00007f69c8b36800 nid=0x73fec waiting on condition [0x00007f6728218000]
"aliyun-log-producer-234-success-batch-handler" #1237 daemon prio=5 os_prio=0 tid=0x00007f69c8b35000 nid=0x73feb waiting on condition [0x00007f6728319000]
"aliyun-log-producer-234-mover" #1236 daemon prio=5 os_prio=0 tid=0x00007f69c8b22000 nid=0x73fea waiting on condition [0x00007f672841a000]
"aliyun-log-producer-233-timeout-thread-0" #1235 daemon prio=5 os_prio=0 tid=0x00007f6b9d4db000 nid=0x73fe9 waiting on condition [0x00007f672851b000]
"aliyun-log-producer-233-io-thread-0" #1234 daemon prio=5 os_prio=0 tid=0x00007f69c8b23000 nid=0x73fe8 waiting on condition [0x00007f672861c000]
"aliyun-log-producer-233-failure-batch-handler" #1233 daemon prio=5 os_prio=0 tid=0x00007f69c8b1e800 nid=0x73fe7 waiting on condition [0x00007f672871d000]
"aliyun-log-producer-233-success-batch-handler" #1232 daemon prio=5 os_prio=0 tid=0x00007f69c8b1d000 nid=0x73fe6 waiting on condition [0x00007f672881e000]
"aliyun-log-producer-233-mover" #1231 daemon prio=5 os_prio=0 tid=0x00007f69c8b1a000 nid=0x73fe5 waiting on condition [0x00007f672891f000]
"aliyun-log-producer-232-timeout-thread-0" #1230 daemon prio=5 os_prio=0 tid=0x00007f6b7c00c800 nid=0x73fe4 waiting on condition [0x00007f6728a20000]
"aliyun-log-producer-232-io-thread-0" #1229 daemon prio=5 os_prio=0 tid=0x00007f69c8b1b000 nid=0x73fe3 waiting on condition [0x00007f6728b21000]
"aliyun-log-producer-232-failure-batch-handler" #1228 daemon prio=5 os_prio=0 tid=0x00007f69c8b16800 nid=0x73fe2 waiting on condition [0x00007f6728c22000]
"aliyun-log-producer-232-success-batch-handler" #1227 daemon prio=5 os_prio=0 tid=0x00007f69c8b15000 nid=0x73fe1 waiting on condition [0x00007f6728d23000]
"aliyun-log-producer-232-mover" #1226 daemon prio=5 os_prio=0 tid=0x00007f69c8b11800 nid=0x73fe0 waiting on condition [0x00007f6728e24000]
"aliyun-log-producer-231-timeout-thread-0" #1225 daemon prio=5 os_prio=0 tid=0x00007f6b60012000 nid=0x73fde waiting on condition [0x00007f6728f25000]
"aliyun-log-producer-231-io-thread-0" #1224 daemon prio=5 os_prio=0 tid=0x00007f69c8b12800 nid=0x73fdd waiting on condition [0x00007f6729026000]
"aliyun-log-producer-231-failure-batch-handler" #1223 daemon prio=5 os_prio=0 tid=0x00007f69c8b0e000 nid=0x73fdc waiting on condition [0x00007f6729127000]
"aliyun-log-producer-231-success-batch-handler" #1222 daemon prio=5 os_prio=0 tid=0x00007f69c8b0c800 nid=0x73fdb waiting on condition [0x00007f6729228000]
"aliyun-log-producer-231-mover" #1221 daemon prio=5 os_prio=0 tid=0x00007f69c899c000 nid=0x73fda waiting on condition [0x00007f6729329000]
"aliyun-log-producer-230-timeout-thread-0" #1220 daemon prio=5 os_prio=0 tid=0x00007f6b54020800 nid=0x73fd9 waiting on condition [0x00007f672962a000]
"aliyun-log-producer-230-io-thread-0" #1219 daemon prio=5 os_prio=0 tid=0x00007f69c8b0a800 nid=0x73fd8 waiting on condition [0x00007f672972b000]
"aliyun-log-producer-230-failure-batch-handler" #1218 daemon prio=5 os_prio=0 tid=0x00007f69c89a9000 nid=0x73fd7 waiting on condition [0x00007f672982c000]
"aliyun-log-producer-230-success-batch-handler" #1217 daemon prio=5 os_prio=0 tid=0x00007f69c89a4800 nid=0x73fd6 waiting on condition [0x00007f672992d000]
"aliyun-log-producer-230-mover" #1216 daemon prio=5 os_prio=0 tid=0x00007f69c8993800 nid=0x73fd5 runnable [0x00007f6729a2e000]
"aliyun-log-producer-229-timeout-thread-0" #1215 daemon prio=5 os_prio=0 tid=0x00007f6b38011000 nid=0x73fd4 waiting on condition [0x00007f6729b2f000]
"aliyun-log-producer-229-io-thread-0" #1214 daemon prio=5 os_prio=0 tid=0x00007f69c89a5800 nid=0x73fd3 waiting on condition [0x00007f6729c30000]
"aliyun-log-producer-229-failure-batch-handler" #1213 daemon prio=5 os_prio=0 tid=0x00007f69c89a1800 nid=0x73fd2 waiting on condition [0x00007f6729d31000]
"aliyun-log-producer-229-success-batch-handler" #1212 daemon prio=5 os_prio=0 tid=0x00007f69c89a1000 nid=0x73fd1 waiting on condition [0x00007f6729e32000]
"aliyun-log-producer-229-mover" #1211 daemon prio=5 os_prio=0 tid=0x00007f69c899f800 nid=0x73fd0 runnable [0x00007f6729f33000]

一直往上涨

Our application can only run on JDK 6 and we found that the guava dependency which was used in the producer was too high version to JDK 6. So we can't init producer successfully in our application.

我在项目中引用了
<dependency> <groupId>com.aliyun.openservices</groupId> <artifactId>aliyun-log</artifactId> <version>0.6.35</version> <classifier>jar-with-dependencies</classifier> </dependency>

其中包含了和apache包完全同名的类
org.apache.commons.collections.CollectionUtils
其中的方法 size 在传入null对象会产生NPE错误，而apache包则不会
这个类和apache的类路径完全相同，难以分辨，希望修一下

aliyun的类

apache的类

missing flush that we've no idea is it ok now to make progress (write checkpoint and handle next batch) after send some data. the only choice seems to close the producer or hold all the ListenableFuture and wait

上图信号量锁释放应该在finally中，catch exception不能catch error。

public List expiredBatches(long timeoutMs) {
long deadline = System.currentTimeMillis() + timeoutMs;
List expiredBatches = new ArrayList();
retryBatches.drainTo(expiredBatches);
if (!expiredBatches.isEmpty()) {
return expiredBatches;
}
while (true) {
if (timeoutMs < 0) {
break;
}
ProducerBatch batch;
try {
batch = retryBatches.poll(timeoutMs, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
LOGGER.info("Interrupted when poll batch from the retry batches");
break;
}
if (batch == null) {
break;
}
expiredBatches.add(batch);
retryBatches.drainTo(expiredBatches);
if (!expiredBatches.isEmpty()) { //这里为什么要判断是不是空？不是上面加了个不为空的batch？
break;
}
timeoutMs = deadline - System.currentTimeMillis(); //按理说走不到这里啊，batch为空直接退出了循环，batch不为空加上了，然后退出返回了？
}
return expiredBatches;
}

maven仓库下载不了0.3.1的jar，0.3.0的可以

com.aliyun.openservices.log.util.NetworkUtils 这个类不见了，是在哪个mvn包下下面

版本信息

<dependency>
	<groupId>com.aliyun.openservices</groupId>
	<artifactId>aliyun-log-producer</artifactId>
	<version>0.3.10</version>
</dependency>
<dependency>
	<groupId>com.aliyun.openservices</groupId>
	<artifactId>aliyun-log</artifactId>
	<version>0.6.35</version>
	<classifier>jar-with-dependencies</classifier>
</dependency>
<dependency>
	<groupId>com.google.protobuf</groupId>
	<artifactId>protobuf-java</artifactId>
	<version>2.5.0</version>
</dependency>

配置信息

total-size-in-bytes: 104857600
max-block-ms: 60000
io-thread-count: 4
batch-size-threshold-in-bytes: 524288
batch-count-threshold: 4096
linger-ms: 2088
retries: 10
base-retry-backoff-ms: 100
max-retry-backoff-ms: 50000

问题描述

1. 在服务产生 OutOfMemoryError: GC overhead limit exceeded 后，Producer发送超时，重试线程异常信息如下：

2023-09-19 13:43:51.096 ERROR 26329 --- [r-3-io-thread-0]
 c.a.o.a.l.p.internals.ProducerBatch      : Failed to put logs, project=demo, logStore=demo-logstore, e=
com.aliyun.openservices.log.exception.LogException: Web request failed: java.lang.IllegalStateException cannot be cast to java.io.IOException
[ErrorCode]: Unknown
[RequestId]: Unknown
	at com.aliyun.openservices.log.Client.SendData(Client.java:2128)
	at com.aliyun.openservices.log.Client.PutLogs(Client.java:660)
	at com.aliyun.openservices.aliyun.log.producer.internals.SendProducerBatchTask.sendProducerBatch(SendProducerBatchTask.java:93)
	at com.aliyun.openservices.aliyun.log.producer.internals.SendProducerBatchTask.run(SendProducerBatchTask.java:62)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
Caused by: com.aliyun.openservices.log.http.client.ClientException: java.lang.IllegalStateException cannot be cast to java.io.IOException
[ErrorCode]: Unknown
[RequestId]: Unknown
	at com.aliyun.openservices.log.http.comm.ServiceClient.sendRequestImpl(ServiceClient.java:93)
	at com.aliyun.openservices.log.http.comm.ServiceClient.sendRequest(ServiceClient.java:67)
	at com.aliyun.openservices.log.Client.SendData(Client.java:2109)
	... 8 common frames omitted
Caused by: java.lang.ClassCastException: java.lang.IllegalStateException cannot be cast to java.io.IOException
	at com.aliyun.openservices.log.http.comm.TimeoutServiceClient.sendRequestCore(TimeoutServiceClient.java:92)
	at com.aliyun.openservices.log.http.comm.ServiceClient.sendRequestImpl(ServiceClient.java:87)
	... 10 common frames omitted

2. 在系统恢复正常后，重试线程仍报错并进入循环抛出异常 (系统日志一直抛出该异常。疑似死循环)：

2023-09-19 13:58:58.020 ERROR 26329 --- [r-3-io-thread-0]
c.a.o.a.l.p.internals.ProducerBatch      : Failed to put logs, project=demo, logStore=demo-logstore, e=

com.aliyun.openservices.log.exception.LogException: Web request failed: null
[ErrorCode]: Unknown
[RequestId]: Unknown
	at com.aliyun.openservices.log.Client.SendData(Client.java:2128)
	at com.aliyun.openservices.log.Client.PutLogs(Client.java:660)
	at com.aliyun.openservices.aliyun.log.producer.internals.SendProducerBatchTask.sendProducerBatch(SendProducerBatchTask.java:93)
	at com.aliyun.openservices.aliyun.log.producer.internals.SendProducerBatchTask.run(SendProducerBatchTask.java:62)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
Caused by: com.aliyun.openservices.log.http.client.ClientException: null
[ErrorCode]: Unknown
[RequestId]: Unknown
	at com.aliyun.openservices.log.http.comm.ServiceClient.sendRequestImpl(ServiceClient.java:93)
	at com.aliyun.openservices.log.http.comm.ServiceClient.sendRequest(ServiceClient.java:67)
	at com.aliyun.openservices.log.Client.SendData(Client.java:2109)
	... 8 common frames omitted
Caused by: java.lang.ClassCastException: null

3. 重试线程循环抛出异常，导致CPU占用过高

Provides transitive vulnerable dependency com.alibaba:fastjson:1.2.83_noneautotype
CVE-2022-25845 9.8 Deserialization of Untrusted Data vulnerability pending CVSS allocation

For current version.

    <dependency>
            <groupId>com.aliyun.openservices</groupId>
            <artifactId>aliyun-log-producer</artifactId>
            <version>0.3.11</version>
     </dependency>

When log producer init, it will call com.aliyun.openservices.aliyun.log.producer.internals.Utils to generateProducerHash. But this method use java.lang.management.ManagementFactory, this is a JDK API, won't be supported in android.

版本

com.aliyun.openservices
aliyun-log-producer
0.3.8


com.aliyun.openservices
aliyun-log
0.6.60


com.google.protobuf
protobuf-java
2.5.0

send ,close, send 就报错了.

Exception in thread "main" java.lang.IllegalStateException: cannot append after the log accumulator was closed
at com.aliyun.openservices.aliyun.log.producer.internals.LogAccumulator.doAppend(LogAccumulator.java:100)
at com.aliyun.openservices.aliyun.log.producer.internals.LogAccumulator.append(LogAccumulator.java:84)
at com.aliyun.openservices.aliyun.log.producer.LogProducer.send(LogProducer.java:385)
at com.aliyun.openservices.aliyun.log.producer.LogProducer.send(LogProducer.java:308)
at com.aliyun.openservices.aliyun.log.producer.LogProducer.send(LogProducer.java:240)
at com.south.aliyun.LogUtil.log(LogUtil.java:83)
at com.south.aliyun.LogUtil.info(LogUtil.java:99)
at com.south.aliyun.LogUtil.info(LogUtil.java:102)
at com.south.aliyun.LogUtil.main(LogUtil.java:134)