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

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

Browse Source
This commit is contained in:
v-zhangjc9
2024-06-05 17:37:32 +08:00
parent 0b4cec3955
commit f0a6dbbbb5
1 changed files with 26 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -138,6 +138,7 @@ public class PulsarMessageSourceReader extends RichParallelSourceFunction<String
public void run(SourceContext<String> context) throws Exception { public void run(SourceContext<String> context) throws Exception {
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();
@@ -169,6 +170,7 @@ public class PulsarMessageSourceReader extends RichParallelSourceFunction<String
} }
} }
} }
}
@Override @Override
public void open(Configuration configuration) throws Exception { public void open(Configuration configuration) throws Exception {