lanyuanxiaoyao/hudi-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

revert(sync): 使用checkpoint lock报障message id上报的时候状态稳定

Browse Source 
会导致部分hudi表checkpoint时间超长
This reverts commit f0a6dbbb
This commit is contained in:
v-zhangjc9
2024-06-17 18:52:01 +08:00
parent 7f10d53c4e
commit 2be10d8437
1 changed files with 26 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
utils/sync/src/main/java/com/lanyuanxiaoyao/service/sync/functions/PulsarMessageSourceReader.java
View File

@@ -123,42 +123,37 @@ public class PulsarMessageSourceReader extends RichParallelSourceFunction<String
.create()) { .create()) {
String currentValue = null; String currentValue = null;
while (running) { while (running) {
synchronized (context.getCheckpointLock()) { Message<String> message;
Message<String> message; try {
try { message = reader.readNext();
message = reader.readNext(); if (ObjectUtil.isNotNull(message)) {
if (ObjectUtil.isNotNull(message)) { String value = message.getValue();
String value = message.getValue(); currentValue = value;
currentValue = value; if (ObjectUtil.isEmpty(value)) {
if (ObjectUtil.isEmpty(value)) { logger.warn("{} {}", message.getValue(), message.getMessageId());
logger.warn("{} {}", message.getValue(), message.getMessageId()); }
} synchronized (context.getCheckpointLock()) {
synchronized (context.getCheckpointLock()) { context.collect(value);
context.collect(value); }
}
if (RecordHelper.isNotVersionUpdateRecord(value)) {
if (RecordHelper.isNotVersionUpdateRecord(value)) { latestPublishTime.set(message.getPublishTime());
latestPublishTime.set(message.getPublishTime()); latestReceiveTime.set(Instant.now().toEpochMilli());
latestReceiveTime.set(Instant.now().toEpochMilli()); }
} lastMessageId.set(message.getMessageId());
lastMessageId.set(message.getMessageId());
messageReceiveCounter.inc();
messageReceiveCounter.inc(); try {
try { messageReceiveBytesCounter.inc(message.getValue().getBytes().length);
messageReceiveBytesCounter.inc(message.getValue().getBytes().length); } catch (Throwable t) {
} catch (Throwable t) { logger.warn("Parse message size failure", t);
logger.warn("Parse message size failure", t);
}
} }
} catch (Throwable t) {
throw new Exception("Read message failure, current value: " + currentValue, t);
} }
} catch (Throwable t) {
throw new Exception("Read message failure, current value: " + currentValue, t);
} }
} }
} }
} catch (Throwable exception) {
logger.error(StrUtil.format("Any error ({})", tableMeta.getAlias()), exception);
throw exception;
} }
} }