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

fix(api): 修复latest_op_ts无法记录

Browse Source 
mysql的greatest函数，null永远是最大值
This commit is contained in:
v-zhangjc9
2024-05-08 14:16:46 +08:00
parent 10fae04007
commit 23ced7a2e1
5 changed files with 35 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
utils/sync/src/main/java/com/lanyuanxiaoyao/service/sync/functions/CompactionEventHandler.java
View File

@@ -40,7 +40,7 @@ public class CompactionEventHandler implements CompactEventHandler, Serializable
public void success(String instant, List<WriteStatus> statuses, HoodieCommitMetadata metadata) {
StatusUtils.compactionCommit(configuration, flinkJob, tableMeta, instant, metadata);
// Trace latest_op_ts
Long max = statuses.stream()
statuses.stream()
.map(status -> {
if (status instanceof TraceWriteStatus) {
TraceWriteStatus s = (TraceWriteStatus) status;
@@ -49,9 +49,10 @@ public class CompactionEventHandler implements CompactEventHandler, Serializable
return 0L;
})
.max(Long::compare)
.orElse(0L);
logger.info("Latest op ts: {}", max);
StatusUtils.compactionLatestOpTs(configuration, flinkJob, tableMeta, max);
.ifPresent(max -> {
logger.info("Latest op ts: {}", max);
StatusUtils.compactionLatestOpTs(configuration, flinkJob, tableMeta, max);
});
}
@Override

6
utils/sync/src/main/java/com/lanyuanxiaoyao/service/sync/utils/StatusUtils.java
View File

@@ -58,7 +58,7 @@ public class StatusUtils {
}
public static void syncCheckpoint(GlobalConfiguration configuration, FlinkJob flinkJob, TableMeta tableMeta, String messageId, Long publishTime) {
logger.info("Enter method: syncCheckpoint[configuration, flinkJob, tableMeta, messageId, publishTime]. " + "configuration:" + configuration + "," + "flinkJob:" + flinkJob + "," + "tableMeta:" + tableMeta + "," + "messageId:" + messageId + "," + "publishTime:" + publishTime);
logger.info("Enter method: syncCheckpoint[configuration, flinkJob, tableMeta, messageId, publishTime]. " + "configuration:" + configuration + "," + "flinkJob:" + flinkJob.getId() + "," + "tableMeta:" + tableMeta.getAlias() + "," + "messageId:" + messageId + "," + "publishTime:" + publishTime);
try {
Failsafe.with(RetryPolicyProvider.HTTP_RETRY)
.run(() -> HttpUtil.createGet(
@@ -229,7 +229,7 @@ public class StatusUtils {
}
public static void versionUpdate(GlobalConfiguration configuration, FlinkJob flinkJob, TableMeta tableMeta, String version, String opts) {
logger.info("Enter method: versionUpdate[configuration, flinkJob, tableMeta, version, opts]. " + "configuration:" + configuration + "," + "flinkJob:" + flinkJob + "," + "tableMeta:" + tableMeta + "," + "version:" + version + "," + "opts:" + opts);
logger.info("Enter method: versionUpdate[configuration, flinkJob, tableMeta, version, opts]. " + "configuration:" + configuration + "," + "flinkJob:" + flinkJob.getId() + "," + "tableMeta:" + tableMeta.getAlias() + "," + "version:" + version + "," + "opts:" + opts);
Failsafe.with(RetryPolicyProvider.HTTP_RETRY)
.run(() ->
HttpUtil.createGet(
@@ -250,7 +250,7 @@ public class StatusUtils {
}
public static void compactionLatestOpTs(GlobalConfiguration configuration, FlinkJob flinkJob, TableMeta tableMeta, Long latestOpTs) {
logger.info("Enter method: compactionLatestOpTs[configuration, flinkJob, tableMeta, latestOpTs]. " + "configuration:" + configuration + "," + "flinkJob:" + flinkJob + "," + "tableMeta:" + tableMeta + "," + "latestOpTs:" + latestOpTs);
logger.info("Enter method: compactionLatestOpTs[configuration, flinkJob, tableMeta, latestOpTs]. " + "configuration:" + configuration + "," + "flinkJob:" + flinkJob.getId() + "," + "tableMeta:" + tableMeta.getAlias() + "," + "latestOpTs:" + latestOpTs);
Failsafe.with(RetryPolicyProvider.HTTP_RETRY)
.run(() ->
HttpUtil.createGet(