wsarecv commented on code in PR #3701:
URL: https://github.com/apache/gobblin/pull/3701#discussion_r1227756280
##########
gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/IcebergMetadataWriter.java:
##########
@@ -836,15 +851,10 @@ public void flush(String dbName, String tableName) throws
IOException {
// The logic is to check the window [currentHour-1,currentHour] and
update the watermark if there are no audit counts
if(!tableMetadata.appendFiles.isPresent() &&
!tableMetadata.deleteFiles.isPresent()
&& tableMetadata.completenessEnabled) {
- if (tableMetadata.completionWatermark >
DEFAULT_COMPLETION_WATERMARK) {
- log.info(String.format("Checking kafka audit for %s on
change_property ", topicName));
- SortedSet<ZonedDateTime> timestamps = new TreeSet<>();
- ZonedDateTime dtAtBeginningOfHour =
ZonedDateTime.now(ZoneId.of(this.timeZone)).truncatedTo(ChronoUnit.HOURS);
- timestamps.add(dtAtBeginningOfHour);
- checkAndUpdateCompletenessWatermark(tableMetadata, topicName,
timestamps, props);
- } else {
- log.info(String.format("Need valid watermark, current watermark is
%s, Not checking kafka audit for %s",
- tableMetadata.completionWatermark, topicName));
+ updateWatermarkWithEmptyFilesRegistered(topicName, tableMetadata,
props, false);
+
+ if (tableMetadata.totalCountCompletenessEnabled) {
+ updateWatermarkWithEmptyFilesRegistered(topicName, tableMetadata,
props, true);
Review Comment:
Fixed
##########
gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/IcebergMetadataWriter.java:
##########
@@ -891,94 +901,45 @@ public void flush(String dbName, String tableName) throws
IOException {
}
}
- @Override
- public void reset(String dbName, String tableName) throws IOException {
- this.tableMetadataMap.remove(TableIdentifier.of(dbName, tableName));
+ private AbstractCompletenessWatermarkUpdater getWatermarkUpdater(String
topicName, TableMetadata tableMetadata,
+ Map<String, String> propsToUpdate, boolean isTotalCountCompleteness) {
+ return isTotalCountCompleteness
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
