You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@gobblin.apache.org by GitBox <gi...@apache.org> on 2022/06/06 20:20:03 UTC
[GitHub] [gobblin] vikrambohra opened a new pull request, #3517: [GOBBLIN-1657] Update completion watermark on change_property in IcebergMetadataWriter
vikrambohra opened a new pull request, #3517:
URL: https://github.com/apache/gobblin/pull/3517
Dear Gobblin maintainers,
Please accept this PR. I understand that it will not be reviewed until I have checked off all the steps below!
### JIRA
- [ ] My PR addresses the following [Gobblin JIRA](https://issues.apache.org/jira/browse/GOBBLIN/) issues and references them in the PR title. For example, "[GOBBLIN-1657] My Gobblin PR"
- https://issues.apache.org/jira/browse/GOBBLIN-1657
### Description
- [ ] Here are some details about my PR, including screenshots (if applicable):
see description in https://issues.apache.org/jira/browse/GOBBLIN-1657
### Tests
- [ ] My PR adds the following unit tests __OR__ does not need testing for this extremely good reason:
### Commits
- [ ] My commits all reference JIRA issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)":
1. Subject is separated from body by a blank line
2. Subject is limited to 50 characters
3. Subject does not end with a period
4. Subject uses the imperative mood ("add", not "adding")
5. Body wraps at 72 characters
6. Body explains "what" and "why", not "how"
--
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: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] sv2000 commented on a diff in pull request #3517: [GOBBLIN-1657] Update completion watermark on change_property in IcebergMetadataWriter
Posted by GitBox <gi...@apache.org>.
sv2000 commented on code in PR #3517:
URL: https://github.com/apache/gobblin/pull/3517#discussion_r890666692
##########
gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/IcebergMetadataWriter.java:
##########
@@ -790,28 +789,33 @@ public void flush(String dbName, String tableName) throws IOException {
Transaction transaction = tableMetadata.transaction.get();
Map<String, String> props = tableMetadata.newProperties.or(
Maps.newHashMap(tableMetadata.lastProperties.or(getIcebergTable(tid).properties())));
+ String topic = props.get(TOPIC_NAME_KEY);
if (tableMetadata.appendFiles.isPresent()) {
tableMetadata.appendFiles.get().commit();
if (tableMetadata.completenessEnabled) {
- String topicName = props.get(TOPIC_NAME_KEY);
- if(topicName == null) {
- log.error(String.format("Not performing audit check. %s is null. Please set as table property of %s.%s",
- TOPIC_NAME_KEY, dbName, tableName));
- } else {
- long newCompletenessWatermark =
- computeCompletenessWatermark(topicName, tableMetadata.datePartitions, tableMetadata.prevCompletenessWatermark);
- if(newCompletenessWatermark > tableMetadata.prevCompletenessWatermark) {
- log.info(String.format("Updating %s for %s.%s to %s", COMPLETION_WATERMARK_KEY, dbName, tableName, newCompletenessWatermark));
- props.put(COMPLETION_WATERMARK_KEY, String.valueOf(newCompletenessWatermark));
- props.put(COMPLETION_WATERMARK_TIMEZONE_KEY, this.timeZone);
- tableMetadata.newCompletenessWatermark = newCompletenessWatermark;
- }
- }
+ checkAndUpdateCompletenessWatermark(tableMetadata, topic, tableMetadata.datePartitions, props);
}
}
if (tableMetadata.deleteFiles.isPresent()) {
tableMetadata.deleteFiles.get().commit();
}
+ // Check and update completion watermark when there are no files to be registered, typically for quiet topics
+ // The logic is to check the next window from previous completion watermark and update the watermark if there are no audit counts
Review Comment:
Is the next window the starting time for completeness checks or will we only check for the completeness of the next window?
##########
gobblin-completeness/src/main/java/org/apache/gobblin/completeness/verifier/KafkaAuditCountVerifier.java:
##########
@@ -45,6 +45,8 @@ public class KafkaAuditCountVerifier {
public static final String REFERENCE_TIERS = COMPLETENESS_PREFIX + "reference.tiers";
public static final String THRESHOLD = COMPLETENESS_PREFIX + "threshold";
private static final double DEFAULT_THRESHOLD = 0.999;
+ public static final String COMPLETE_ON_NO_COUNTS = COMPLETENESS_PREFIX + "complete.on.no.counts";
Review Comment:
Do we need a separate config? Shouldn't we always move completion watermark for quiet topics?
--
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: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] sv2000 commented on a diff in pull request #3517: [GOBBLIN-1657] Update completion watermark on change_property in IcebergMetadataWriter
Posted by GitBox <gi...@apache.org>.
sv2000 commented on code in PR #3517:
URL: https://github.com/apache/gobblin/pull/3517#discussion_r891531637
##########
gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/IcebergMetadataWriter.java:
##########
@@ -790,28 +789,33 @@ public void flush(String dbName, String tableName) throws IOException {
Transaction transaction = tableMetadata.transaction.get();
Map<String, String> props = tableMetadata.newProperties.or(
Maps.newHashMap(tableMetadata.lastProperties.or(getIcebergTable(tid).properties())));
+ String topic = props.get(TOPIC_NAME_KEY);
if (tableMetadata.appendFiles.isPresent()) {
tableMetadata.appendFiles.get().commit();
if (tableMetadata.completenessEnabled) {
- String topicName = props.get(TOPIC_NAME_KEY);
- if(topicName == null) {
- log.error(String.format("Not performing audit check. %s is null. Please set as table property of %s.%s",
- TOPIC_NAME_KEY, dbName, tableName));
- } else {
- long newCompletenessWatermark =
- computeCompletenessWatermark(topicName, tableMetadata.datePartitions, tableMetadata.prevCompletenessWatermark);
- if(newCompletenessWatermark > tableMetadata.prevCompletenessWatermark) {
- log.info(String.format("Updating %s for %s.%s to %s", COMPLETION_WATERMARK_KEY, dbName, tableName, newCompletenessWatermark));
- props.put(COMPLETION_WATERMARK_KEY, String.valueOf(newCompletenessWatermark));
- props.put(COMPLETION_WATERMARK_TIMEZONE_KEY, this.timeZone);
- tableMetadata.newCompletenessWatermark = newCompletenessWatermark;
- }
- }
+ checkAndUpdateCompletenessWatermark(tableMetadata, topic, tableMetadata.datePartitions, props);
}
}
if (tableMetadata.deleteFiles.isPresent()) {
tableMetadata.deleteFiles.get().commit();
}
+ // Check and update completion watermark when there are no files to be registered, typically for quiet topics
+ // The logic is to check the next window from previous completion watermark and update the watermark if there are no audit counts
Review Comment:
So what happens if the current time is 08:00 and current watermark is 02:00? Will we only check for the completeness for [02:00, 03:00]? Shouldn't we check for completeness of the entire range [02:00, 08:00]?
--
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: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] sv2000 merged pull request #3517: [GOBBLIN-1657] Update completion watermark on change_property in IcebergMetadataWriter
Posted by GitBox <gi...@apache.org>.
sv2000 merged PR #3517:
URL: https://github.com/apache/gobblin/pull/3517
--
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: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] codecov-commenter commented on pull request #3517: [GOBBLIN-1657] Update completion watermark on change_property in IcebergMetadataWriter
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #3517:
URL: https://github.com/apache/gobblin/pull/3517#issuecomment-1147899665
# [Codecov](https://codecov.io/gh/apache/gobblin/pull/3517?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#3517](https://codecov.io/gh/apache/gobblin/pull/3517?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (eaf6116) into [master](https://codecov.io/gh/apache/gobblin/commit/7b29ce2911e5d77410c313f71d315dc8bc82cf5b?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (7b29ce2) will **increase** coverage by `4.67%`.
> The diff coverage is `n/a`.
```diff
@@ Coverage Diff @@
## master #3517 +/- ##
============================================
+ Coverage 46.71% 51.39% +4.67%
+ Complexity 10426 6919 -3507
============================================
Files 2080 1294 -786
Lines 81305 48273 -33032
Branches 9072 5401 -3671
============================================
- Hits 37985 24809 -13176
+ Misses 39825 21272 -18553
+ Partials 3495 2192 -1303
```
| [Impacted Files](https://codecov.io/gh/apache/gobblin/pull/3517?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...gobblin/service/modules/core/GitConfigMonitor.java](https://codecov.io/gh/apache/gobblin/pull/3517/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1zZXJ2aWNlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3NlcnZpY2UvbW9kdWxlcy9jb3JlL0dpdENvbmZpZ01vbml0b3IuamF2YQ==) | `81.35% <0.00%> (-1.70%)` | :arrow_down: |
| [...ata/management/copy/hive/HiveCopyEntityHelper.java](https://codecov.io/gh/apache/gobblin/pull/3517/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1kYXRhLW1hbmFnZW1lbnQvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2dvYmJsaW4vZGF0YS9tYW5hZ2VtZW50L2NvcHkvaGl2ZS9IaXZlQ29weUVudGl0eUhlbHBlci5qYXZh) | `62.14% <0.00%> (-0.85%)` | :arrow_down: |
| [...anagement/copy/replication/ConfigBasedDataset.java](https://codecov.io/gh/apache/gobblin/pull/3517/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1kYXRhLW1hbmFnZW1lbnQvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2dvYmJsaW4vZGF0YS9tYW5hZ2VtZW50L2NvcHkvcmVwbGljYXRpb24vQ29uZmlnQmFzZWREYXRhc2V0LmphdmE=) | `68.87% <0.00%> (ø)` | |
| [...completeness/verifier/KafkaAuditCountVerifier.java](https://codecov.io/gh/apache/gobblin/pull/3517/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1jb21wbGV0ZW5lc3Mvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2dvYmJsaW4vY29tcGxldGVuZXNzL3ZlcmlmaWVyL0thZmthQXVkaXRDb3VudFZlcmlmaWVyLmphdmE=) | | |
| [...gobblin/iceberg/publisher/GobblinMCEPublisher.java](https://codecov.io/gh/apache/gobblin/pull/3517/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvcHVibGlzaGVyL0dvYmJsaW5NQ0VQdWJsaXNoZXIuamF2YQ==) | | |
| [.../gobblin/iceberg/writer/IcebergMetadataWriter.java](https://codecov.io/gh/apache/gobblin/pull/3517/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvd3JpdGVyL0ljZWJlcmdNZXRhZGF0YVdyaXRlci5qYXZh) | | |
| [...pache/gobblin/metrics/kafka/KafkaAvroReporter.java](https://codecov.io/gh/apache/gobblin/pull/3517/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1tb2R1bGVzL2dvYmJsaW4ta2Fma2EtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL21ldHJpY3Mva2Fma2EvS2Fma2FBdnJvUmVwb3J0ZXIuamF2YQ==) | | |
| [...gestion/google/webmaster/UrlTriePrefixGrouper.java](https://codecov.io/gh/apache/gobblin/pull/3517/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1tb2R1bGVzL2dvb2dsZS1pbmdlc3Rpb24vc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2dvYmJsaW4vaW5nZXN0aW9uL2dvb2dsZS93ZWJtYXN0ZXIvVXJsVHJpZVByZWZpeEdyb3VwZXIuamF2YQ==) | | |
| [...zedRecordToEncryptedSerializedRecordConverter.java](https://codecov.io/gh/apache/gobblin/pull/3517/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1tb2R1bGVzL2dvYmJsaW4tY3J5cHRvLXByb3ZpZGVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NvbnZlcnRlci9TZXJpYWxpemVkUmVjb3JkVG9FbmNyeXB0ZWRTZXJpYWxpemVkUmVjb3JkQ29udmVydGVyLmphdmE=) | | |
| [...org/apache/gobblin/crypto/HexKeyToStringCodec.java](https://codecov.io/gh/apache/gobblin/pull/3517/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1tb2R1bGVzL2dvYmJsaW4tY3J5cHRvL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NyeXB0by9IZXhLZXlUb1N0cmluZ0NvZGVjLmphdmE=) | | |
| ... and [782 more](https://codecov.io/gh/apache/gobblin/pull/3517/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/gobblin/pull/3517?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/gobblin/pull/3517?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [7b29ce2...eaf6116](https://codecov.io/gh/apache/gobblin/pull/3517?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] sv2000 commented on pull request #3517: [GOBBLIN-1657] Update completion watermark on change_property in IcebergMetadataWriter
Posted by GitBox <gi...@apache.org>.
sv2000 commented on PR #3517:
URL: https://github.com/apache/gobblin/pull/3517#issuecomment-1148068678
@vikrambohra Let's include PR description here instead of linking to an Apache JIRA. In addition, is there any chance we can add some unit tests?
--
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: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] sv2000 commented on pull request #3517: [GOBBLIN-1657] Update completion watermark on change_property in IcebergMetadataWriter
Posted by GitBox <gi...@apache.org>.
sv2000 commented on PR #3517:
URL: https://github.com/apache/gobblin/pull/3517#issuecomment-1148068031
@vikrambohra Also check for travis build failure.
--
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: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] vikrambohra commented on a diff in pull request #3517: [GOBBLIN-1657] Update completion watermark on change_property in IcebergMetadataWriter
Posted by GitBox <gi...@apache.org>.
vikrambohra commented on code in PR #3517:
URL: https://github.com/apache/gobblin/pull/3517#discussion_r890699926
##########
gobblin-completeness/src/main/java/org/apache/gobblin/completeness/verifier/KafkaAuditCountVerifier.java:
##########
@@ -45,6 +45,8 @@ public class KafkaAuditCountVerifier {
public static final String REFERENCE_TIERS = COMPLETENESS_PREFIX + "reference.tiers";
public static final String THRESHOLD = COMPLETENESS_PREFIX + "threshold";
private static final double DEFAULT_THRESHOLD = 0.999;
+ public static final String COMPLETE_ON_NO_COUNTS = COMPLETENESS_PREFIX + "complete.on.no.counts";
Review Comment:
Default behavior is for other users of this class to not want a "success" if there is an empty response from kafka audit.
This config is only for this use case where we want a "success" on empty response from kafka audit.
--
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: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] vikrambohra commented on a diff in pull request #3517: [GOBBLIN-1657] Update completion watermark on change_property in IcebergMetadataWriter
Posted by GitBox <gi...@apache.org>.
vikrambohra commented on code in PR #3517:
URL: https://github.com/apache/gobblin/pull/3517#discussion_r891629386
##########
gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/IcebergMetadataWriter.java:
##########
@@ -790,28 +789,33 @@ public void flush(String dbName, String tableName) throws IOException {
Transaction transaction = tableMetadata.transaction.get();
Map<String, String> props = tableMetadata.newProperties.or(
Maps.newHashMap(tableMetadata.lastProperties.or(getIcebergTable(tid).properties())));
+ String topic = props.get(TOPIC_NAME_KEY);
if (tableMetadata.appendFiles.isPresent()) {
tableMetadata.appendFiles.get().commit();
if (tableMetadata.completenessEnabled) {
- String topicName = props.get(TOPIC_NAME_KEY);
- if(topicName == null) {
- log.error(String.format("Not performing audit check. %s is null. Please set as table property of %s.%s",
- TOPIC_NAME_KEY, dbName, tableName));
- } else {
- long newCompletenessWatermark =
- computeCompletenessWatermark(topicName, tableMetadata.datePartitions, tableMetadata.prevCompletenessWatermark);
- if(newCompletenessWatermark > tableMetadata.prevCompletenessWatermark) {
- log.info(String.format("Updating %s for %s.%s to %s", COMPLETION_WATERMARK_KEY, dbName, tableName, newCompletenessWatermark));
- props.put(COMPLETION_WATERMARK_KEY, String.valueOf(newCompletenessWatermark));
- props.put(COMPLETION_WATERMARK_TIMEZONE_KEY, this.timeZone);
- tableMetadata.newCompletenessWatermark = newCompletenessWatermark;
- }
- }
+ checkAndUpdateCompletenessWatermark(tableMetadata, topic, tableMetadata.datePartitions, props);
}
}
if (tableMetadata.deleteFiles.isPresent()) {
tableMetadata.deleteFiles.get().commit();
}
+ // Check and update completion watermark when there are no files to be registered, typically for quiet topics
+ // The logic is to check the next window from previous completion watermark and update the watermark if there are no audit counts
Review Comment:
This case should happen only after a long downtime. Even then watermark should catchup quickly as we will be flushing every minute. And if there is a append file operation in between then we would catchup more aggressively
In regular scenario. We will update watermark on the next flush once current time is the next hour of current watermark.
--
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: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] vikrambohra commented on a diff in pull request #3517: [GOBBLIN-1657] Update completion watermark on change_property in IcebergMetadataWriter
Posted by GitBox <gi...@apache.org>.
vikrambohra commented on code in PR #3517:
URL: https://github.com/apache/gobblin/pull/3517#discussion_r890688489
##########
gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/IcebergMetadataWriter.java:
##########
@@ -790,28 +789,33 @@ public void flush(String dbName, String tableName) throws IOException {
Transaction transaction = tableMetadata.transaction.get();
Map<String, String> props = tableMetadata.newProperties.or(
Maps.newHashMap(tableMetadata.lastProperties.or(getIcebergTable(tid).properties())));
+ String topic = props.get(TOPIC_NAME_KEY);
if (tableMetadata.appendFiles.isPresent()) {
tableMetadata.appendFiles.get().commit();
if (tableMetadata.completenessEnabled) {
- String topicName = props.get(TOPIC_NAME_KEY);
- if(topicName == null) {
- log.error(String.format("Not performing audit check. %s is null. Please set as table property of %s.%s",
- TOPIC_NAME_KEY, dbName, tableName));
- } else {
- long newCompletenessWatermark =
- computeCompletenessWatermark(topicName, tableMetadata.datePartitions, tableMetadata.prevCompletenessWatermark);
- if(newCompletenessWatermark > tableMetadata.prevCompletenessWatermark) {
- log.info(String.format("Updating %s for %s.%s to %s", COMPLETION_WATERMARK_KEY, dbName, tableName, newCompletenessWatermark));
- props.put(COMPLETION_WATERMARK_KEY, String.valueOf(newCompletenessWatermark));
- props.put(COMPLETION_WATERMARK_TIMEZONE_KEY, this.timeZone);
- tableMetadata.newCompletenessWatermark = newCompletenessWatermark;
- }
- }
+ checkAndUpdateCompletenessWatermark(tableMetadata, topic, tableMetadata.datePartitions, props);
}
}
if (tableMetadata.deleteFiles.isPresent()) {
tableMetadata.deleteFiles.get().commit();
}
+ // Check and update completion watermark when there are no files to be registered, typically for quiet topics
+ // The logic is to check the next window from previous completion watermark and update the watermark if there are no audit counts
Review Comment:
Just the next window = [currentWatermark, currentWatermark + 1 hour]. However audit service will not be called if this window falls in the current hour.
--
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: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org