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 2021/10/25 21:32:04 UTC
[GitHub] [gobblin] ZihanLi58 opened a new pull request #3419: [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container
ZihanLi58 opened a new pull request #3419:
URL: https://github.com/apache/gobblin/pull/3419
…e will not affect other topics in the same container
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-XXX] My Gobblin PR"
- https://issues.apache.org/jira/browse/GOBBLIN-1565
### Description
- [ ] Here are some details about my PR, including screenshots (if applicable):
GMIP should be fault tolerant, so that one topic failure should not affect other topics in the same container. One potential solution is that we can catch the error and document the error dataset, if we have more than one datasets have issue, that normally something wrong happening on the whole job, we should fail the task, but if there is only one problematic topic, we can handle the exception and let the pipeline move on and process other topics.
Left some todo for emitting GTE so that we can rely on those event to detect failure and do backfill as needed
### Tests
- [ ] My PR adds the following unit tests __OR__ does not need testing for this extremely good reason:
Add new unit test
### 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] codecov-commenter edited a comment on pull request #3419: [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #3419:
URL: https://github.com/apache/gobblin/pull/3419#issuecomment-951356726
# [Codecov](https://codecov.io/gh/apache/gobblin/pull/3419?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 [#3419](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (2786777) into [master](https://codecov.io/gh/apache/gobblin/commit/255bdc15f4e8dd7dbf5fb94e9742d8430f2506f3?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (255bdc1) will **decrease** coverage by `3.21%`.
> The diff coverage is `78.20%`.
[](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #3419 +/- ##
============================================
- Coverage 46.53% 43.31% -3.22%
+ Complexity 10278 2021 -8257
============================================
Files 2065 402 -1663
Lines 80638 17325 -63313
Branches 9005 2121 -6884
============================================
- Hits 37521 7505 -30016
+ Misses 39641 8982 -30659
+ Partials 3476 838 -2638
```
| [Impacted Files](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...pache/gobblin/iceberg/writer/GobblinMCEWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvd3JpdGVyL0dvYmJsaW5NQ0VXcml0ZXIuamF2YQ==) | `74.70% <73.84%> (+2.34%)` | :arrow_up: |
| [...bblin/iceberg/writer/GobblinMetadataException.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvd3JpdGVyL0dvYmJsaW5NZXRhZGF0YUV4Y2VwdGlvbi5qYXZh) | `100.00% <100.00%> (ø)` | |
| [.../gobblin/iceberg/writer/IcebergMetadataWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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) | `72.29% <100.00%> (+0.18%)` | :arrow_up: |
| [.../java/org/apache/gobblin/cluster/SleepingTask.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1jbHVzdGVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NsdXN0ZXIvU2xlZXBpbmdUYXNrLmphdmE=) | `39.39% <0.00%> (-6.07%)` | :arrow_down: |
| [...main/java/org/apache/gobblin/util/HadoopUtils.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi11dGlsaXR5L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3V0aWwvSGFkb29wVXRpbHMuamF2YQ==) | `29.93% <0.00%> (-0.66%)` | :arrow_down: |
| [.../apache/gobblin/cluster/GobblinClusterManager.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1jbHVzdGVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NsdXN0ZXIvR29iYmxpbkNsdXN0ZXJNYW5hZ2VyLmphdmE=) | `53.27% <0.00%> (-0.65%)` | :arrow_down: |
| [...apache/gobblin/salesforce/SalesforceExtractor.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1zYWxlc2ZvcmNlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3NhbGVzZm9yY2UvU2FsZXNmb3JjZUV4dHJhY3Rvci5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...apache/gobblin/hive/writer/HiveMetadataWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1oaXZlLXJlZ2lzdHJhdGlvbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZ29iYmxpbi9oaXZlL3dyaXRlci9IaXZlTWV0YWRhdGFXcml0ZXIuamF2YQ==) | | |
| [...ction/mapreduce/CompactionAvroJobConfigurator.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1jb21wYWN0aW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NvbXBhY3Rpb24vbWFwcmVkdWNlL0NvbXBhY3Rpb25BdnJvSm9iQ29uZmlndXJhdG9yLmphdmE=) | | |
| ... and [1666 more](https://codecov.io/gh/apache/gobblin/pull/3419/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/3419?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/3419?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 [255bdc1...2786777](https://codecov.io/gh/apache/gobblin/pull/3419?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] codecov-commenter edited a comment on pull request #3419: [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #3419:
URL: https://github.com/apache/gobblin/pull/3419#issuecomment-951356726
# [Codecov](https://codecov.io/gh/apache/gobblin/pull/3419?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 [#3419](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (2786777) into [master](https://codecov.io/gh/apache/gobblin/commit/255bdc15f4e8dd7dbf5fb94e9742d8430f2506f3?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (255bdc1) will **increase** coverage by `1.88%`.
> The diff coverage is `72.61%`.
[](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #3419 +/- ##
============================================
+ Coverage 46.53% 48.41% +1.88%
+ Complexity 10278 7685 -2593
============================================
Files 2065 1437 -628
Lines 80638 56760 -23878
Branches 9005 6536 -2469
============================================
- Hits 37521 27483 -10038
+ Misses 39641 26710 -12931
+ Partials 3476 2567 -909
```
| [Impacted Files](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...apache/gobblin/hive/writer/HiveMetadataWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1oaXZlLXJlZ2lzdHJhdGlvbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZ29iYmxpbi9oaXZlL3dyaXRlci9IaXZlTWV0YWRhdGFXcml0ZXIuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...pache/gobblin/iceberg/writer/GobblinMCEWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvd3JpdGVyL0dvYmJsaW5NQ0VXcml0ZXIuamF2YQ==) | `74.70% <73.84%> (+2.34%)` | :arrow_up: |
| [...bblin/iceberg/writer/GobblinMetadataException.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvd3JpdGVyL0dvYmJsaW5NZXRhZGF0YUV4Y2VwdGlvbi5qYXZh) | `100.00% <100.00%> (ø)` | |
| [.../gobblin/iceberg/writer/IcebergMetadataWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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) | `72.29% <100.00%> (+0.18%)` | :arrow_up: |
| [.../java/org/apache/gobblin/cluster/SleepingTask.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1jbHVzdGVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NsdXN0ZXIvU2xlZXBpbmdUYXNrLmphdmE=) | `39.39% <0.00%> (-6.07%)` | :arrow_down: |
| [...compaction/verify/CompactionTimeRangeVerifier.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1jb21wYWN0aW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NvbXBhY3Rpb24vdmVyaWZ5L0NvbXBhY3Rpb25UaW1lUmFuZ2VWZXJpZmllci5qYXZh) | `86.88% <0.00%> (-1.64%)` | :arrow_down: |
| [.../apache/gobblin/runtime/api/JobExecutionState.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3J1bnRpbWUvYXBpL0pvYkV4ZWN1dGlvblN0YXRlLmphdmE=) | `79.43% <0.00%> (-0.94%)` | :arrow_down: |
| [...main/java/org/apache/gobblin/yarn/YarnService.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi15YXJuL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3lhcm4vWWFyblNlcnZpY2UuamF2YQ==) | `15.46% <0.00%> (-0.78%)` | :arrow_down: |
| [...main/java/org/apache/gobblin/util/HadoopUtils.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi11dGlsaXR5L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3V0aWwvSGFkb29wVXRpbHMuamF2YQ==) | `29.93% <0.00%> (-0.66%)` | :arrow_down: |
| [.../apache/gobblin/cluster/GobblinClusterManager.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1jbHVzdGVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NsdXN0ZXIvR29iYmxpbkNsdXN0ZXJNYW5hZ2VyLmphdmE=) | `53.27% <0.00%> (-0.65%)` | :arrow_down: |
| ... and [651 more](https://codecov.io/gh/apache/gobblin/pull/3419/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/3419?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/3419?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 [255bdc1...2786777](https://codecov.io/gh/apache/gobblin/pull/3419?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] codecov-commenter edited a comment on pull request #3419: [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #3419:
URL: https://github.com/apache/gobblin/pull/3419#issuecomment-951356726
# [Codecov](https://codecov.io/gh/apache/gobblin/pull/3419?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 [#3419](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (c6bcb1c) into [master](https://codecov.io/gh/apache/gobblin/commit/255bdc15f4e8dd7dbf5fb94e9742d8430f2506f3?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (255bdc1) will **increase** coverage by `2.58%`.
> The diff coverage is `73.80%`.
[](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #3419 +/- ##
============================================
+ Coverage 46.53% 49.11% +2.58%
+ Complexity 10278 8849 -1429
============================================
Files 2065 1690 -375
Lines 80638 65280 -15358
Branches 9005 7482 -1523
============================================
- Hits 37521 32062 -5459
+ Misses 39641 30189 -9452
+ Partials 3476 3029 -447
```
| [Impacted Files](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...apache/gobblin/hive/writer/HiveMetadataWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1oaXZlLXJlZ2lzdHJhdGlvbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZ29iYmxpbi9oaXZlL3dyaXRlci9IaXZlTWV0YWRhdGFXcml0ZXIuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...pache/gobblin/iceberg/writer/GobblinMCEWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvd3JpdGVyL0dvYmJsaW5NQ0VXcml0ZXIuamF2YQ==) | `74.01% <75.75%> (+1.65%)` | :arrow_up: |
| [...bblin/iceberg/writer/GobblinMetadataException.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvd3JpdGVyL0dvYmJsaW5NZXRhZGF0YUV4Y2VwdGlvbi5qYXZh) | `100.00% <100.00%> (ø)` | |
| [.../gobblin/iceberg/writer/IcebergMetadataWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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) | `72.21% <100.00%> (+0.10%)` | :arrow_up: |
| [.../java/org/apache/gobblin/cluster/SleepingTask.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1jbHVzdGVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NsdXN0ZXIvU2xlZXBpbmdUYXNrLmphdmE=) | `39.39% <0.00%> (-6.07%)` | :arrow_down: |
| [...lin/restli/throttling/ZookeeperLeaderElection.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1yZXN0bGkvZ29iYmxpbi10aHJvdHRsaW5nLXNlcnZpY2UvZ29iYmxpbi10aHJvdHRsaW5nLXNlcnZpY2Utc2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3Jlc3RsaS90aHJvdHRsaW5nL1pvb2tlZXBlckxlYWRlckVsZWN0aW9uLmphdmE=) | `70.00% <0.00%> (-2.23%)` | :arrow_down: |
| [...main/java/org/apache/gobblin/yarn/YarnService.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi15YXJuL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3lhcm4vWWFyblNlcnZpY2UuamF2YQ==) | `15.46% <0.00%> (-0.78%)` | :arrow_down: |
| [...pache/gobblin/configuration/ConfigurationKeys.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1hcGkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2dvYmJsaW4vY29uZmlndXJhdGlvbi9Db25maWd1cmF0aW9uS2V5cy5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...apache/gobblin/salesforce/SalesforceExtractor.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1zYWxlc2ZvcmNlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3NhbGVzZm9yY2UvU2FsZXNmb3JjZUV4dHJhY3Rvci5qYXZh) | `0.00% <0.00%> (ø)` | |
| ... and [393 more](https://codecov.io/gh/apache/gobblin/pull/3419/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/3419?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/3419?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 [255bdc1...c6bcb1c](https://codecov.io/gh/apache/gobblin/pull/3419?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 a change in pull request #3419: [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container
Posted by GitBox <gi...@apache.org>.
sv2000 commented on a change in pull request #3419:
URL: https://github.com/apache/gobblin/pull/3419#discussion_r745842141
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -344,12 +347,13 @@ private void flush(String dbName, String tableName) throws IOException {
writer.flush(dbName, tableName);
} catch (IOException e) {
meetException = true;
- addException(e, tableString, dbName, tableName);
+ writer.reset(dbName, tableName);
+ addOrThrowException(e, tableString, dbName, tableName);
}
}
}
- if (!meetException && datasetErrorMap.containsKey(tableOperationTypeMap.get(tableString).datasetName)
- && datasetErrorMap.get(tableOperationTypeMap.get(tableString).datasetName).containsKey(tableString)) {
+ if (!meetException && datasetErrorMap.containsKey(tableOperationTypeMap.get(tableString).datasetPath)
Review comment:
So the if condition is hard to understand. What is the scenario this is trying to address? Can we add a comment here to help the reader understand?
--
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 change in pull request #3419: [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container
Posted by GitBox <gi...@apache.org>.
sv2000 commented on a change in pull request #3419:
URL: https://github.com/apache/gobblin/pull/3419#discussion_r740443463
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -248,20 +272,88 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
String dbName = spec.getTable().getDbName();
String tableName = spec.getTable().getTableName();
String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
- if (tableOperationTypeMap.containsKey(tableString)
- && tableOperationTypeMap.get(tableString) != gmce.getOperationType()) {
- for (MetadataWriter writer : metadataWriters) {
- writer.flush(dbName, tableName);
- }
- }
- tableOperationTypeMap.put(tableString, gmce.getOperationType());
- for (MetadataWriter writer : metadataWriters) {
- writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ if (!tableOperationTypeMap.containsKey(tableString)) {
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
+ } else if (tableOperationTypeMap.get(tableString).operationType != gmce.getOperationType()) {
+ flush(dbName, tableName);
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
}
+ tableOperationTypeMap.get(tableString).gmceHighWatermark = watermark.getLwm().getValue();
+ write(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
}
this.recordCount.incrementAndGet();
}
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void write(RecordEnvelope recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
+ boolean meetException = false;
+ String dbName = spec.getTable().getDbName();
+ String tableName = spec.getTable().getTableName();
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ for (MetadataWriter writer : metadataWriters) {
+ if (meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ } catch (Exception e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ }
+ private void addException(Exception e, String tableString, String dbName, String tableName) throws IOException{
Review comment:
Add new line before the start of a new method.
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -248,20 +272,88 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
String dbName = spec.getTable().getDbName();
String tableName = spec.getTable().getTableName();
String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
- if (tableOperationTypeMap.containsKey(tableString)
- && tableOperationTypeMap.get(tableString) != gmce.getOperationType()) {
- for (MetadataWriter writer : metadataWriters) {
- writer.flush(dbName, tableName);
- }
- }
- tableOperationTypeMap.put(tableString, gmce.getOperationType());
- for (MetadataWriter writer : metadataWriters) {
- writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ if (!tableOperationTypeMap.containsKey(tableString)) {
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
+ } else if (tableOperationTypeMap.get(tableString).operationType != gmce.getOperationType()) {
+ flush(dbName, tableName);
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
}
+ tableOperationTypeMap.get(tableString).gmceHighWatermark = watermark.getLwm().getValue();
+ write(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
}
this.recordCount.incrementAndGet();
}
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void write(RecordEnvelope recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
+ boolean meetException = false;
+ String dbName = spec.getTable().getDbName();
+ String tableName = spec.getTable().getTableName();
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ for (MetadataWriter writer : metadataWriters) {
+ if (meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ } catch (Exception e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ }
+ private void addException(Exception e, String tableString, String dbName, String tableName) throws IOException{
+ TableStatus tableStatus = tableOperationTypeMap.get(tableString);
+ Map<String, Queue<MetadataFlushException>> tableErrorMap = this.datasetErrorMap.getOrDefault(tableStatus.datasetName, new HashMap<>());
+ Queue<MetadataFlushException> errors = tableErrorMap.getOrDefault(tableString, new LinkedList<>());
+ if (!errors.isEmpty() && errors.peek().highWatermark == tableStatus.gmceLowWatermark) {
+ errors.peek().highWatermark = tableStatus.gmceHighWatermark;
+ } else {
+ MetadataFlushException metadataFlushException =
+ new MetadataFlushException(tableStatus.datasetName, dbName, tableName, tableStatus.gmceTopicName, tableStatus.gmceTopicPartition, tableStatus.gmceLowWatermark, tableStatus.gmceHighWatermark, e);
+ errors.offer(metadataFlushException);
+ }
+ tableErrorMap.put(tableString, errors);
+ this.datasetErrorMap.put(tableStatus.datasetName, tableErrorMap);
Review comment:
This is confusing. How is datasetName different from tableString? If there is a 1:1 mapping between datasetName and tableString, is Map of Maps the right data structure here?
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -187,6 +210,7 @@ public Descriptor getDataDescriptor() {
@Override
public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws IOException {
GenericRecord genericRecord = recordEnvelope.getRecord();
+ KafkaStreamingExtractor.KafkaWatermark watermark = ((KafkaStreamingExtractor.KafkaWatermark) recordEnvelope.getWatermark());
Review comment:
Looks like we are leaking details here. Ideally, we should not expose details of extractor implementation here. Can we use the CheckpointableWatermark interface instead?
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/MetadataFlushException.java
##########
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.iceberg.writer;
+
+import java.io.IOException;
+
+
+public class MetadataFlushException extends IOException {
+ public String datasetName;
+ public String dbName;
+ public String tableName;
+ public String GMCETopicName;
+ public int GMCETopicPartition;
+ public long highWatermark;
+ public long lowWatermark;
+ public Exception exception;
+ MetadataFlushException(String datasetName, String dbName, String tableName, String GMCETopicName,
Review comment:
Since this is used in both write and flush, shall we call it GMCEException or something more generic?
##########
File path: gobblin-hive-registration/src/main/java/org/apache/gobblin/hive/writer/MetadataWriter.java
##########
@@ -47,6 +47,16 @@
*/
void flush(String dbName, String tableName) throws IOException;
+ /**
+ * If something wrong happens, we want to reset the table inside the writer so that we can continual
Review comment:
continual -> continue
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -248,20 +272,88 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
String dbName = spec.getTable().getDbName();
String tableName = spec.getTable().getTableName();
String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
- if (tableOperationTypeMap.containsKey(tableString)
- && tableOperationTypeMap.get(tableString) != gmce.getOperationType()) {
- for (MetadataWriter writer : metadataWriters) {
- writer.flush(dbName, tableName);
- }
- }
- tableOperationTypeMap.put(tableString, gmce.getOperationType());
- for (MetadataWriter writer : metadataWriters) {
- writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ if (!tableOperationTypeMap.containsKey(tableString)) {
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
+ } else if (tableOperationTypeMap.get(tableString).operationType != gmce.getOperationType()) {
+ flush(dbName, tableName);
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
}
+ tableOperationTypeMap.get(tableString).gmceHighWatermark = watermark.getLwm().getValue();
+ write(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
}
this.recordCount.incrementAndGet();
}
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void write(RecordEnvelope recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
+ boolean meetException = false;
+ String dbName = spec.getTable().getDbName();
+ String tableName = spec.getTable().getTableName();
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ for (MetadataWriter writer : metadataWriters) {
+ if (meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ } catch (Exception e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ }
+ private void addException(Exception e, String tableString, String dbName, String tableName) throws IOException{
+ TableStatus tableStatus = tableOperationTypeMap.get(tableString);
+ Map<String, Queue<MetadataFlushException>> tableErrorMap = this.datasetErrorMap.getOrDefault(tableStatus.datasetName, new HashMap<>());
+ Queue<MetadataFlushException> errors = tableErrorMap.getOrDefault(tableString, new LinkedList<>());
+ if (!errors.isEmpty() && errors.peek().highWatermark == tableStatus.gmceLowWatermark) {
+ errors.peek().highWatermark = tableStatus.gmceHighWatermark;
+ } else {
+ MetadataFlushException metadataFlushException =
+ new MetadataFlushException(tableStatus.datasetName, dbName, tableName, tableStatus.gmceTopicName, tableStatus.gmceTopicPartition, tableStatus.gmceLowWatermark, tableStatus.gmceHighWatermark, e);
+ errors.offer(metadataFlushException);
+ }
+ tableErrorMap.put(tableString, errors);
+ this.datasetErrorMap.put(tableStatus.datasetName, tableErrorMap);
+ log.error(String.format("Meet exception when flush table %s", tableString), e);
+ if (datasetErrorMap.size() > maxErrorDataset) {
+ //Fail the job if the error size exceeds some number
+ throw new IOException(String.format("Container fails to flush for more than %s dataset", maxErrorDataset));
+ }
+ tableStatus.gmceLowWatermark = tableStatus.gmceHighWatermark;
+ }
+
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void flush(String dbName, String tableName) throws IOException {
+ boolean meetException = false;
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ if (tableOperationTypeMap.get(tableString).gmceLowWatermark == tableOperationTypeMap.get(tableString).gmceHighWatermark) {
+ // No need to flush
+ return;
+ }
+ for (MetadataWriter writer : metadataWriters) {
+ if(meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.flush(dbName, tableName);
+ } catch (IOException e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ if (!meetException && datasetErrorMap.containsKey(tableOperationTypeMap.get(tableString).datasetName)
Review comment:
Shouldn't this if condition be triggered if meetException = true?
##########
File path: gobblin-hive-registration/src/main/java/org/apache/gobblin/hive/writer/MetadataWriter.java
##########
@@ -47,6 +47,16 @@
*/
void flush(String dbName, String tableName) throws IOException;
+ /**
+ * If something wrong happens, we want to reset the table inside the writer so that we can continual
Review comment:
The meaning of reset is not quite clear from the javadoc description. Is it about cleaning up some in-memory state that is maintained by the pipeline for the table?
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -248,20 +272,88 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
String dbName = spec.getTable().getDbName();
String tableName = spec.getTable().getTableName();
String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
- if (tableOperationTypeMap.containsKey(tableString)
- && tableOperationTypeMap.get(tableString) != gmce.getOperationType()) {
- for (MetadataWriter writer : metadataWriters) {
- writer.flush(dbName, tableName);
- }
- }
- tableOperationTypeMap.put(tableString, gmce.getOperationType());
- for (MetadataWriter writer : metadataWriters) {
- writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ if (!tableOperationTypeMap.containsKey(tableString)) {
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
+ } else if (tableOperationTypeMap.get(tableString).operationType != gmce.getOperationType()) {
+ flush(dbName, tableName);
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
}
+ tableOperationTypeMap.get(tableString).gmceHighWatermark = watermark.getLwm().getValue();
+ write(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
}
this.recordCount.incrementAndGet();
}
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void write(RecordEnvelope recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
+ boolean meetException = false;
+ String dbName = spec.getTable().getDbName();
+ String tableName = spec.getTable().getTableName();
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ for (MetadataWriter writer : metadataWriters) {
+ if (meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ } catch (Exception e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ }
+ private void addException(Exception e, String tableString, String dbName, String tableName) throws IOException{
+ TableStatus tableStatus = tableOperationTypeMap.get(tableString);
+ Map<String, Queue<MetadataFlushException>> tableErrorMap = this.datasetErrorMap.getOrDefault(tableStatus.datasetName, new HashMap<>());
+ Queue<MetadataFlushException> errors = tableErrorMap.getOrDefault(tableString, new LinkedList<>());
+ if (!errors.isEmpty() && errors.peek().highWatermark == tableStatus.gmceLowWatermark) {
+ errors.peek().highWatermark = tableStatus.gmceHighWatermark;
+ } else {
+ MetadataFlushException metadataFlushException =
+ new MetadataFlushException(tableStatus.datasetName, dbName, tableName, tableStatus.gmceTopicName, tableStatus.gmceTopicPartition, tableStatus.gmceLowWatermark, tableStatus.gmceHighWatermark, e);
+ errors.offer(metadataFlushException);
+ }
+ tableErrorMap.put(tableString, errors);
+ this.datasetErrorMap.put(tableStatus.datasetName, tableErrorMap);
+ log.error(String.format("Meet exception when flush table %s", tableString), e);
+ if (datasetErrorMap.size() > maxErrorDataset) {
+ //Fail the job if the error size exceeds some number
+ throw new IOException(String.format("Container fails to flush for more than %s dataset", maxErrorDataset));
Review comment:
Should we throw the actual exception that the flush/write encountered for the specific dataset?
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -248,20 +272,88 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
String dbName = spec.getTable().getDbName();
String tableName = spec.getTable().getTableName();
String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
- if (tableOperationTypeMap.containsKey(tableString)
- && tableOperationTypeMap.get(tableString) != gmce.getOperationType()) {
- for (MetadataWriter writer : metadataWriters) {
- writer.flush(dbName, tableName);
- }
- }
- tableOperationTypeMap.put(tableString, gmce.getOperationType());
- for (MetadataWriter writer : metadataWriters) {
- writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ if (!tableOperationTypeMap.containsKey(tableString)) {
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
+ } else if (tableOperationTypeMap.get(tableString).operationType != gmce.getOperationType()) {
+ flush(dbName, tableName);
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
}
+ tableOperationTypeMap.get(tableString).gmceHighWatermark = watermark.getLwm().getValue();
+ write(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
}
this.recordCount.incrementAndGet();
}
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void write(RecordEnvelope recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
+ boolean meetException = false;
+ String dbName = spec.getTable().getDbName();
+ String tableName = spec.getTable().getTableName();
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ for (MetadataWriter writer : metadataWriters) {
+ if (meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ } catch (Exception e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ }
+ private void addException(Exception e, String tableString, String dbName, String tableName) throws IOException{
Review comment:
Should we call this method addOrThrowException?
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/MetadataFlushException.java
##########
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.iceberg.writer;
+
+import java.io.IOException;
+
+
+public class MetadataFlushException extends IOException {
+ public String datasetName;
Review comment:
Seems like there is a lot of overlap in the fields: datasetName, dbName, tableName, topicName, topicPartition. Can we keep only the ones needed and derive the rest from those fields?
--
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 edited a comment on pull request #3419: [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #3419:
URL: https://github.com/apache/gobblin/pull/3419#issuecomment-951356726
# [Codecov](https://codecov.io/gh/apache/gobblin/pull/3419?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 [#3419](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (1a8a00a) into [master](https://codecov.io/gh/apache/gobblin/commit/255bdc15f4e8dd7dbf5fb94e9742d8430f2506f3?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (255bdc1) will **increase** coverage by `0.01%`.
> The diff coverage is `71.25%`.
[](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #3419 +/- ##
============================================
+ Coverage 46.53% 46.54% +0.01%
- Complexity 10278 10288 +10
============================================
Files 2065 2066 +1
Lines 80638 80706 +68
Branches 9005 9011 +6
============================================
+ Hits 37521 37564 +43
- Misses 39641 39662 +21
- Partials 3476 3480 +4
```
| [Impacted Files](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...apache/gobblin/hive/writer/HiveMetadataWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1oaXZlLXJlZ2lzdHJhdGlvbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZ29iYmxpbi9oaXZlL3dyaXRlci9IaXZlTWV0YWRhdGFXcml0ZXIuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [.../gobblin/iceberg/writer/IcebergMetadataWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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) | `71.84% <0.00%> (-0.27%)` | :arrow_down: |
| [...pache/gobblin/iceberg/writer/GobblinMCEWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvd3JpdGVyL0dvYmJsaW5NQ0VXcml0ZXIuamF2YQ==) | `74.13% <76.19%> (+1.78%)` | :arrow_up: |
| [...gobblin/iceberg/writer/MetadataFlushException.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvd3JpdGVyL01ldGFkYXRhRmx1c2hFeGNlcHRpb24uamF2YQ==) | `100.00% <100.00%> (ø)` | |
| [.../java/org/apache/gobblin/cluster/SleepingTask.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1jbHVzdGVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NsdXN0ZXIvU2xlZXBpbmdUYXNrLmphdmE=) | `39.39% <0.00%> (-6.07%)` | :arrow_down: |
| [...lin/restli/throttling/ZookeeperLeaderElection.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1yZXN0bGkvZ29iYmxpbi10aHJvdHRsaW5nLXNlcnZpY2UvZ29iYmxpbi10aHJvdHRsaW5nLXNlcnZpY2Utc2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3Jlc3RsaS90aHJvdHRsaW5nL1pvb2tlZXBlckxlYWRlckVsZWN0aW9uLmphdmE=) | `70.00% <0.00%> (-2.23%)` | :arrow_down: |
| [...a/management/copy/publisher/CopyDataPublisher.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1kYXRhLW1hbmFnZW1lbnQvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2dvYmJsaW4vZGF0YS9tYW5hZ2VtZW50L2NvcHkvcHVibGlzaGVyL0NvcHlEYXRhUHVibGlzaGVyLmphdmE=) | `74.17% <0.00%> (-1.33%)` | :arrow_down: |
| [.../apache/gobblin/runtime/api/JobExecutionState.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3J1bnRpbWUvYXBpL0pvYkV4ZWN1dGlvblN0YXRlLmphdmE=) | `79.43% <0.00%> (-0.94%)` | :arrow_down: |
| [...main/java/org/apache/gobblin/yarn/YarnService.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi15YXJuL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3lhcm4vWWFyblNlcnZpY2UuamF2YQ==) | `15.46% <0.00%> (-0.78%)` | :arrow_down: |
| [...main/java/org/apache/gobblin/util/HadoopUtils.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi11dGlsaXR5L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3V0aWwvSGFkb29wVXRpbHMuamF2YQ==) | `29.93% <0.00%> (-0.66%)` | :arrow_down: |
| ... and [4 more](https://codecov.io/gh/apache/gobblin/pull/3419/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/3419?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/3419?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 [255bdc1...1a8a00a](https://codecov.io/gh/apache/gobblin/pull/3419?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] codecov-commenter edited a comment on pull request #3419: [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #3419:
URL: https://github.com/apache/gobblin/pull/3419#issuecomment-951356726
# [Codecov](https://codecov.io/gh/apache/gobblin/pull/3419?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 [#3419](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (c6bcb1c) into [master](https://codecov.io/gh/apache/gobblin/commit/255bdc15f4e8dd7dbf5fb94e9742d8430f2506f3?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (255bdc1) will **decrease** coverage by `3.20%`.
> The diff coverage is `79.48%`.
[](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #3419 +/- ##
============================================
- Coverage 46.53% 43.33% -3.21%
+ Complexity 10278 2018 -8260
============================================
Files 2065 402 -1663
Lines 80638 17339 -63299
Branches 9005 2127 -6878
============================================
- Hits 37521 7513 -30008
+ Misses 39641 8983 -30658
+ Partials 3476 843 -2633
```
| [Impacted Files](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...pache/gobblin/iceberg/writer/GobblinMCEWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvd3JpdGVyL0dvYmJsaW5NQ0VXcml0ZXIuamF2YQ==) | `74.01% <75.75%> (+1.65%)` | :arrow_up: |
| [...bblin/iceberg/writer/GobblinMetadataException.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvd3JpdGVyL0dvYmJsaW5NZXRhZGF0YUV4Y2VwdGlvbi5qYXZh) | `100.00% <100.00%> (ø)` | |
| [.../gobblin/iceberg/writer/IcebergMetadataWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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) | `72.21% <100.00%> (+0.10%)` | :arrow_up: |
| [.../java/org/apache/gobblin/cluster/SleepingTask.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1jbHVzdGVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NsdXN0ZXIvU2xlZXBpbmdUYXNrLmphdmE=) | `39.39% <0.00%> (-6.07%)` | :arrow_down: |
| [...apache/gobblin/salesforce/SalesforceExtractor.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1zYWxlc2ZvcmNlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3NhbGVzZm9yY2UvU2FsZXNmb3JjZUV4dHJhY3Rvci5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...apache/gobblin/hive/writer/HiveMetadataWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1oaXZlLXJlZ2lzdHJhdGlvbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZ29iYmxpbi9oaXZlL3dyaXRlci9IaXZlTWV0YWRhdGFXcml0ZXIuamF2YQ==) | | |
| [...apache/gobblin/writer/GobblinOrcMemoryManager.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1tb2R1bGVzL2dvYmJsaW4tb3JjL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3dyaXRlci9Hb2JibGluT3JjTWVtb3J5TWFuYWdlci5qYXZh) | | |
| [...tadata/provider/SimpleMetadataProviderFactory.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1tb2R1bGVzL2dvYmJsaW4tbWV0YWRhdGEvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2dvYmJsaW4vbWV0YWRhdGEvcHJvdmlkZXIvU2ltcGxlTWV0YWRhdGFQcm92aWRlckZhY3RvcnkuamF2YQ==) | | |
| [...ion/hive/materializer/CopyTableQueryGenerator.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1kYXRhLW1hbmFnZW1lbnQvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2dvYmJsaW4vZGF0YS9tYW5hZ2VtZW50L2NvbnZlcnNpb24vaGl2ZS9tYXRlcmlhbGl6ZXIvQ29weVRhYmxlUXVlcnlHZW5lcmF0b3IuamF2YQ==) | | |
| ... and [1661 more](https://codecov.io/gh/apache/gobblin/pull/3419/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/3419?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/3419?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 [255bdc1...c6bcb1c](https://codecov.io/gh/apache/gobblin/pull/3419?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] codecov-commenter edited a comment on pull request #3419: [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #3419:
URL: https://github.com/apache/gobblin/pull/3419#issuecomment-951356726
--
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 edited a comment on pull request #3419: [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #3419:
URL: https://github.com/apache/gobblin/pull/3419#issuecomment-951356726
--
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] ZihanLi58 commented on pull request #3419: [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container
Posted by GitBox <gi...@apache.org>.
ZihanLi58 commented on pull request #3419:
URL: https://github.com/apache/gobblin/pull/3419#issuecomment-951349379
@sv2000 @vikrambohra Can you help take a look whenever you have time?
--
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 #3419: [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on pull request #3419:
URL: https://github.com/apache/gobblin/pull/3419#issuecomment-951356726
# [Codecov](https://codecov.io/gh/apache/gobblin/pull/3419?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 [#3419](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (1a8a00a) into [master](https://codecov.io/gh/apache/gobblin/commit/255bdc15f4e8dd7dbf5fb94e9742d8430f2506f3?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (255bdc1) will **decrease** coverage by `3.19%`.
> The diff coverage is `77.02%`.
[](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #3419 +/- ##
============================================
- Coverage 46.53% 43.33% -3.20%
+ Complexity 10278 2018 -8260
============================================
Files 2065 402 -1663
Lines 80638 17332 -63306
Branches 9005 2127 -6878
============================================
- Hits 37521 7510 -30011
+ Misses 39641 8980 -30661
+ Partials 3476 842 -2634
```
| [Impacted Files](https://codecov.io/gh/apache/gobblin/pull/3419?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/iceberg/writer/IcebergMetadataWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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) | `71.84% <0.00%> (-0.27%)` | :arrow_down: |
| [...pache/gobblin/iceberg/writer/GobblinMCEWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvd3JpdGVyL0dvYmJsaW5NQ0VXcml0ZXIuamF2YQ==) | `74.13% <76.19%> (+1.78%)` | :arrow_up: |
| [...gobblin/iceberg/writer/MetadataFlushException.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvd3JpdGVyL01ldGFkYXRhRmx1c2hFeGNlcHRpb24uamF2YQ==) | `100.00% <100.00%> (ø)` | |
| [.../java/org/apache/gobblin/cluster/SleepingTask.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1jbHVzdGVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NsdXN0ZXIvU2xlZXBpbmdUYXNrLmphdmE=) | `39.39% <0.00%> (-6.07%)` | :arrow_down: |
| [...main/java/org/apache/gobblin/util/HadoopUtils.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi11dGlsaXR5L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3V0aWwvSGFkb29wVXRpbHMuamF2YQ==) | `29.93% <0.00%> (-0.66%)` | :arrow_down: |
| [...apache/gobblin/hive/writer/HiveMetadataWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1oaXZlLXJlZ2lzdHJhdGlvbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZ29iYmxpbi9oaXZlL3dyaXRlci9IaXZlTWV0YWRhdGFXcml0ZXIuamF2YQ==) | | |
| [.../destination/DestinationDatasetHandlerFactory.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1jb3JlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2Rlc3RpbmF0aW9uL0Rlc3RpbmF0aW9uRGF0YXNldEhhbmRsZXJGYWN0b3J5LmphdmE=) | | |
| [...xtractor/extract/google/GoogleDriveFileSystem.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1tb2R1bGVzL2dvb2dsZS1pbmdlc3Rpb24vc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2dvYmJsaW4vc291cmNlL2V4dHJhY3Rvci9leHRyYWN0L2dvb2dsZS9Hb29nbGVEcml2ZUZpbGVTeXN0ZW0uamF2YQ==) | | |
| [...rg/apache/gobblin/runtime/FsDatasetStateStore.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3J1bnRpbWUvRnNEYXRhc2V0U3RhdGVTdG9yZS5qYXZh) | | |
| ... and [1663 more](https://codecov.io/gh/apache/gobblin/pull/3419/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/3419?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/3419?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 [255bdc1...1a8a00a](https://codecov.io/gh/apache/gobblin/pull/3419?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 merged pull request #3419: [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container
Posted by GitBox <gi...@apache.org>.
sv2000 merged pull request #3419:
URL: https://github.com/apache/gobblin/pull/3419
--
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 edited a comment on pull request #3419: [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #3419:
URL: https://github.com/apache/gobblin/pull/3419#issuecomment-951356726
# [Codecov](https://codecov.io/gh/apache/gobblin/pull/3419?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 [#3419](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (2786777) into [master](https://codecov.io/gh/apache/gobblin/commit/255bdc15f4e8dd7dbf5fb94e9742d8430f2506f3?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (255bdc1) will **increase** coverage by `2.58%`.
> The diff coverage is `72.61%`.
[](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #3419 +/- ##
============================================
+ Coverage 46.53% 49.11% +2.58%
+ Complexity 10278 8852 -1426
============================================
Files 2065 1690 -375
Lines 80638 65264 -15374
Branches 9005 7476 -1529
============================================
- Hits 37521 32055 -5466
+ Misses 39641 30184 -9457
+ Partials 3476 3025 -451
```
| [Impacted Files](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...apache/gobblin/hive/writer/HiveMetadataWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1oaXZlLXJlZ2lzdHJhdGlvbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZ29iYmxpbi9oaXZlL3dyaXRlci9IaXZlTWV0YWRhdGFXcml0ZXIuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...pache/gobblin/iceberg/writer/GobblinMCEWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvd3JpdGVyL0dvYmJsaW5NQ0VXcml0ZXIuamF2YQ==) | `74.70% <73.84%> (+2.34%)` | :arrow_up: |
| [...bblin/iceberg/writer/GobblinMetadataException.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvd3JpdGVyL0dvYmJsaW5NZXRhZGF0YUV4Y2VwdGlvbi5qYXZh) | `100.00% <100.00%> (ø)` | |
| [.../gobblin/iceberg/writer/IcebergMetadataWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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) | `72.29% <100.00%> (+0.18%)` | :arrow_up: |
| [.../java/org/apache/gobblin/cluster/SleepingTask.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1jbHVzdGVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NsdXN0ZXIvU2xlZXBpbmdUYXNrLmphdmE=) | `39.39% <0.00%> (-6.07%)` | :arrow_down: |
| [...compaction/verify/CompactionTimeRangeVerifier.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1jb21wYWN0aW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NvbXBhY3Rpb24vdmVyaWZ5L0NvbXBhY3Rpb25UaW1lUmFuZ2VWZXJpZmllci5qYXZh) | `86.88% <0.00%> (-1.64%)` | :arrow_down: |
| [.../apache/gobblin/runtime/api/JobExecutionState.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3J1bnRpbWUvYXBpL0pvYkV4ZWN1dGlvblN0YXRlLmphdmE=) | `79.43% <0.00%> (-0.94%)` | :arrow_down: |
| [...main/java/org/apache/gobblin/yarn/YarnService.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi15YXJuL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3lhcm4vWWFyblNlcnZpY2UuamF2YQ==) | `15.46% <0.00%> (-0.78%)` | :arrow_down: |
| [...main/java/org/apache/gobblin/util/HadoopUtils.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi11dGlsaXR5L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3V0aWwvSGFkb29wVXRpbHMuamF2YQ==) | `29.93% <0.00%> (-0.66%)` | :arrow_down: |
| [.../apache/gobblin/cluster/GobblinClusterManager.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1jbHVzdGVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NsdXN0ZXIvR29iYmxpbkNsdXN0ZXJNYW5hZ2VyLmphdmE=) | `53.27% <0.00%> (-0.65%)` | :arrow_down: |
| ... and [402 more](https://codecov.io/gh/apache/gobblin/pull/3419/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/3419?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/3419?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 [255bdc1...2786777](https://codecov.io/gh/apache/gobblin/pull/3419?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] ZihanLi58 commented on a change in pull request #3419: [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container
Posted by GitBox <gi...@apache.org>.
ZihanLi58 commented on a change in pull request #3419:
URL: https://github.com/apache/gobblin/pull/3419#discussion_r742452232
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -248,20 +272,88 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
String dbName = spec.getTable().getDbName();
String tableName = spec.getTable().getTableName();
String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
- if (tableOperationTypeMap.containsKey(tableString)
- && tableOperationTypeMap.get(tableString) != gmce.getOperationType()) {
- for (MetadataWriter writer : metadataWriters) {
- writer.flush(dbName, tableName);
- }
- }
- tableOperationTypeMap.put(tableString, gmce.getOperationType());
- for (MetadataWriter writer : metadataWriters) {
- writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ if (!tableOperationTypeMap.containsKey(tableString)) {
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
+ } else if (tableOperationTypeMap.get(tableString).operationType != gmce.getOperationType()) {
+ flush(dbName, tableName);
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
}
+ tableOperationTypeMap.get(tableString).gmceHighWatermark = watermark.getLwm().getValue();
+ write(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
}
this.recordCount.incrementAndGet();
}
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void write(RecordEnvelope recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
+ boolean meetException = false;
+ String dbName = spec.getTable().getDbName();
+ String tableName = spec.getTable().getTableName();
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ for (MetadataWriter writer : metadataWriters) {
+ if (meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ } catch (Exception e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ }
+ private void addException(Exception e, String tableString, String dbName, String tableName) throws IOException{
+ TableStatus tableStatus = tableOperationTypeMap.get(tableString);
+ Map<String, Queue<MetadataFlushException>> tableErrorMap = this.datasetErrorMap.getOrDefault(tableStatus.datasetName, new HashMap<>());
+ Queue<MetadataFlushException> errors = tableErrorMap.getOrDefault(tableString, new LinkedList<>());
+ if (!errors.isEmpty() && errors.peek().highWatermark == tableStatus.gmceLowWatermark) {
+ errors.peek().highWatermark = tableStatus.gmceHighWatermark;
+ } else {
+ MetadataFlushException metadataFlushException =
+ new MetadataFlushException(tableStatus.datasetName, dbName, tableName, tableStatus.gmceTopicName, tableStatus.gmceTopicPartition, tableStatus.gmceLowWatermark, tableStatus.gmceHighWatermark, e);
+ errors.offer(metadataFlushException);
+ }
+ tableErrorMap.put(tableString, errors);
+ this.datasetErrorMap.put(tableStatus.datasetName, tableErrorMap);
+ log.error(String.format("Meet exception when flush table %s", tableString), e);
+ if (datasetErrorMap.size() > maxErrorDataset) {
+ //Fail the job if the error size exceeds some number
+ throw new IOException(String.format("Container fails to flush for more than %s dataset", maxErrorDataset));
Review comment:
I print out the exception also we will emit GTE for those exception. But will update the exception to contain the last exception we met
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -248,20 +272,88 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
String dbName = spec.getTable().getDbName();
String tableName = spec.getTable().getTableName();
String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
- if (tableOperationTypeMap.containsKey(tableString)
- && tableOperationTypeMap.get(tableString) != gmce.getOperationType()) {
- for (MetadataWriter writer : metadataWriters) {
- writer.flush(dbName, tableName);
- }
- }
- tableOperationTypeMap.put(tableString, gmce.getOperationType());
- for (MetadataWriter writer : metadataWriters) {
- writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ if (!tableOperationTypeMap.containsKey(tableString)) {
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
+ } else if (tableOperationTypeMap.get(tableString).operationType != gmce.getOperationType()) {
+ flush(dbName, tableName);
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
}
+ tableOperationTypeMap.get(tableString).gmceHighWatermark = watermark.getLwm().getValue();
+ write(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
}
this.recordCount.incrementAndGet();
}
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void write(RecordEnvelope recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
+ boolean meetException = false;
+ String dbName = spec.getTable().getDbName();
+ String tableName = spec.getTable().getTableName();
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ for (MetadataWriter writer : metadataWriters) {
+ if (meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ } catch (Exception e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ }
+ private void addException(Exception e, String tableString, String dbName, String tableName) throws IOException{
+ TableStatus tableStatus = tableOperationTypeMap.get(tableString);
+ Map<String, Queue<MetadataFlushException>> tableErrorMap = this.datasetErrorMap.getOrDefault(tableStatus.datasetName, new HashMap<>());
+ Queue<MetadataFlushException> errors = tableErrorMap.getOrDefault(tableString, new LinkedList<>());
+ if (!errors.isEmpty() && errors.peek().highWatermark == tableStatus.gmceLowWatermark) {
+ errors.peek().highWatermark = tableStatus.gmceHighWatermark;
+ } else {
+ MetadataFlushException metadataFlushException =
+ new MetadataFlushException(tableStatus.datasetName, dbName, tableName, tableStatus.gmceTopicName, tableStatus.gmceTopicPartition, tableStatus.gmceLowWatermark, tableStatus.gmceHighWatermark, e);
+ errors.offer(metadataFlushException);
+ }
+ tableErrorMap.put(tableString, errors);
+ this.datasetErrorMap.put(tableStatus.datasetName, tableErrorMap);
+ log.error(String.format("Meet exception when flush table %s", tableString), e);
+ if (datasetErrorMap.size() > maxErrorDataset) {
+ //Fail the job if the error size exceeds some number
+ throw new IOException(String.format("Container fails to flush for more than %s dataset", maxErrorDataset));
+ }
+ tableStatus.gmceLowWatermark = tableStatus.gmceHighWatermark;
+ }
+
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void flush(String dbName, String tableName) throws IOException {
+ boolean meetException = false;
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ if (tableOperationTypeMap.get(tableString).gmceLowWatermark == tableOperationTypeMap.get(tableString).gmceHighWatermark) {
+ // No need to flush
+ return;
+ }
+ for (MetadataWriter writer : metadataWriters) {
+ if(meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.flush(dbName, tableName);
+ } catch (IOException e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ if (!meetException && datasetErrorMap.containsKey(tableOperationTypeMap.get(tableString).datasetName)
Review comment:
We want to emit GTE only when the actual watermark moves. it's either in the flush() method where watermark in state store move or we successfully commit one snapshot so that the high watermark in the iceberg table property moves. That's why the condition here is meetException is false as this means we successfully commit once for this table after we met exception. Another place we will emit GTE is at the end of the flush() (not flush(db,table)) method.
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -248,20 +272,88 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
String dbName = spec.getTable().getDbName();
String tableName = spec.getTable().getTableName();
String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
- if (tableOperationTypeMap.containsKey(tableString)
- && tableOperationTypeMap.get(tableString) != gmce.getOperationType()) {
- for (MetadataWriter writer : metadataWriters) {
- writer.flush(dbName, tableName);
- }
- }
- tableOperationTypeMap.put(tableString, gmce.getOperationType());
- for (MetadataWriter writer : metadataWriters) {
- writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ if (!tableOperationTypeMap.containsKey(tableString)) {
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
+ } else if (tableOperationTypeMap.get(tableString).operationType != gmce.getOperationType()) {
+ flush(dbName, tableName);
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
}
+ tableOperationTypeMap.get(tableString).gmceHighWatermark = watermark.getLwm().getValue();
+ write(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
}
this.recordCount.incrementAndGet();
}
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void write(RecordEnvelope recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
+ boolean meetException = false;
+ String dbName = spec.getTable().getDbName();
+ String tableName = spec.getTable().getTableName();
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ for (MetadataWriter writer : metadataWriters) {
+ if (meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ } catch (Exception e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ }
+ private void addException(Exception e, String tableString, String dbName, String tableName) throws IOException{
+ TableStatus tableStatus = tableOperationTypeMap.get(tableString);
+ Map<String, Queue<MetadataFlushException>> tableErrorMap = this.datasetErrorMap.getOrDefault(tableStatus.datasetName, new HashMap<>());
+ Queue<MetadataFlushException> errors = tableErrorMap.getOrDefault(tableString, new LinkedList<>());
+ if (!errors.isEmpty() && errors.peek().highWatermark == tableStatus.gmceLowWatermark) {
+ errors.peek().highWatermark = tableStatus.gmceHighWatermark;
+ } else {
+ MetadataFlushException metadataFlushException =
+ new MetadataFlushException(tableStatus.datasetName, dbName, tableName, tableStatus.gmceTopicName, tableStatus.gmceTopicPartition, tableStatus.gmceLowWatermark, tableStatus.gmceHighWatermark, e);
+ errors.offer(metadataFlushException);
+ }
+ tableErrorMap.put(tableString, errors);
+ this.datasetErrorMap.put(tableStatus.datasetName, tableErrorMap);
Review comment:
Sorry for that confusing, I was using datasetName to be consistent with the metadata schema where they treat path as datasetName. I will change datasetName to datasetPath to avoid this confusion. It's not a 1:1 mapping, as one dataset can be register into multi tables
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -248,20 +272,88 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
String dbName = spec.getTable().getDbName();
String tableName = spec.getTable().getTableName();
String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
- if (tableOperationTypeMap.containsKey(tableString)
- && tableOperationTypeMap.get(tableString) != gmce.getOperationType()) {
- for (MetadataWriter writer : metadataWriters) {
- writer.flush(dbName, tableName);
- }
- }
- tableOperationTypeMap.put(tableString, gmce.getOperationType());
- for (MetadataWriter writer : metadataWriters) {
- writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ if (!tableOperationTypeMap.containsKey(tableString)) {
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
+ } else if (tableOperationTypeMap.get(tableString).operationType != gmce.getOperationType()) {
+ flush(dbName, tableName);
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
}
+ tableOperationTypeMap.get(tableString).gmceHighWatermark = watermark.getLwm().getValue();
+ write(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
}
this.recordCount.incrementAndGet();
}
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void write(RecordEnvelope recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
+ boolean meetException = false;
+ String dbName = spec.getTable().getDbName();
+ String tableName = spec.getTable().getTableName();
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ for (MetadataWriter writer : metadataWriters) {
+ if (meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ } catch (Exception e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ }
+ private void addException(Exception e, String tableString, String dbName, String tableName) throws IOException{
+ TableStatus tableStatus = tableOperationTypeMap.get(tableString);
+ Map<String, Queue<MetadataFlushException>> tableErrorMap = this.datasetErrorMap.getOrDefault(tableStatus.datasetName, new HashMap<>());
+ Queue<MetadataFlushException> errors = tableErrorMap.getOrDefault(tableString, new LinkedList<>());
+ if (!errors.isEmpty() && errors.peek().highWatermark == tableStatus.gmceLowWatermark) {
+ errors.peek().highWatermark = tableStatus.gmceHighWatermark;
+ } else {
+ MetadataFlushException metadataFlushException =
+ new MetadataFlushException(tableStatus.datasetName, dbName, tableName, tableStatus.gmceTopicName, tableStatus.gmceTopicPartition, tableStatus.gmceLowWatermark, tableStatus.gmceHighWatermark, e);
+ errors.offer(metadataFlushException);
+ }
+ tableErrorMap.put(tableString, errors);
+ this.datasetErrorMap.put(tableStatus.datasetName, tableErrorMap);
+ log.error(String.format("Meet exception when flush table %s", tableString), e);
+ if (datasetErrorMap.size() > maxErrorDataset) {
+ //Fail the job if the error size exceeds some number
+ throw new IOException(String.format("Container fails to flush for more than %s dataset", maxErrorDataset));
Review comment:
I print out the exception also we will emit GTE for those exception. But will update the exception to contain the last exception we met
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -248,20 +272,88 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
String dbName = spec.getTable().getDbName();
String tableName = spec.getTable().getTableName();
String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
- if (tableOperationTypeMap.containsKey(tableString)
- && tableOperationTypeMap.get(tableString) != gmce.getOperationType()) {
- for (MetadataWriter writer : metadataWriters) {
- writer.flush(dbName, tableName);
- }
- }
- tableOperationTypeMap.put(tableString, gmce.getOperationType());
- for (MetadataWriter writer : metadataWriters) {
- writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ if (!tableOperationTypeMap.containsKey(tableString)) {
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
+ } else if (tableOperationTypeMap.get(tableString).operationType != gmce.getOperationType()) {
+ flush(dbName, tableName);
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
}
+ tableOperationTypeMap.get(tableString).gmceHighWatermark = watermark.getLwm().getValue();
+ write(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
}
this.recordCount.incrementAndGet();
}
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void write(RecordEnvelope recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
+ boolean meetException = false;
+ String dbName = spec.getTable().getDbName();
+ String tableName = spec.getTable().getTableName();
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ for (MetadataWriter writer : metadataWriters) {
+ if (meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ } catch (Exception e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ }
+ private void addException(Exception e, String tableString, String dbName, String tableName) throws IOException{
+ TableStatus tableStatus = tableOperationTypeMap.get(tableString);
+ Map<String, Queue<MetadataFlushException>> tableErrorMap = this.datasetErrorMap.getOrDefault(tableStatus.datasetName, new HashMap<>());
+ Queue<MetadataFlushException> errors = tableErrorMap.getOrDefault(tableString, new LinkedList<>());
+ if (!errors.isEmpty() && errors.peek().highWatermark == tableStatus.gmceLowWatermark) {
+ errors.peek().highWatermark = tableStatus.gmceHighWatermark;
+ } else {
+ MetadataFlushException metadataFlushException =
+ new MetadataFlushException(tableStatus.datasetName, dbName, tableName, tableStatus.gmceTopicName, tableStatus.gmceTopicPartition, tableStatus.gmceLowWatermark, tableStatus.gmceHighWatermark, e);
+ errors.offer(metadataFlushException);
+ }
+ tableErrorMap.put(tableString, errors);
+ this.datasetErrorMap.put(tableStatus.datasetName, tableErrorMap);
+ log.error(String.format("Meet exception when flush table %s", tableString), e);
+ if (datasetErrorMap.size() > maxErrorDataset) {
+ //Fail the job if the error size exceeds some number
+ throw new IOException(String.format("Container fails to flush for more than %s dataset", maxErrorDataset));
+ }
+ tableStatus.gmceLowWatermark = tableStatus.gmceHighWatermark;
+ }
+
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void flush(String dbName, String tableName) throws IOException {
+ boolean meetException = false;
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ if (tableOperationTypeMap.get(tableString).gmceLowWatermark == tableOperationTypeMap.get(tableString).gmceHighWatermark) {
+ // No need to flush
+ return;
+ }
+ for (MetadataWriter writer : metadataWriters) {
+ if(meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.flush(dbName, tableName);
+ } catch (IOException e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ if (!meetException && datasetErrorMap.containsKey(tableOperationTypeMap.get(tableString).datasetName)
Review comment:
We want to emit GTE only when the actual watermark moves. it's either in the flush() method where watermark in state store move or we successfully commit one snapshot so that the high watermark in the iceberg table property moves. That's why the condition here is meetException is false as this means we successfully commit once for this table after we met exception. Another place we will emit GTE is at the end of the flush() (not flush(db,table)) method.
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -248,20 +272,88 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
String dbName = spec.getTable().getDbName();
String tableName = spec.getTable().getTableName();
String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
- if (tableOperationTypeMap.containsKey(tableString)
- && tableOperationTypeMap.get(tableString) != gmce.getOperationType()) {
- for (MetadataWriter writer : metadataWriters) {
- writer.flush(dbName, tableName);
- }
- }
- tableOperationTypeMap.put(tableString, gmce.getOperationType());
- for (MetadataWriter writer : metadataWriters) {
- writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ if (!tableOperationTypeMap.containsKey(tableString)) {
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
+ } else if (tableOperationTypeMap.get(tableString).operationType != gmce.getOperationType()) {
+ flush(dbName, tableName);
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
}
+ tableOperationTypeMap.get(tableString).gmceHighWatermark = watermark.getLwm().getValue();
+ write(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
}
this.recordCount.incrementAndGet();
}
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void write(RecordEnvelope recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
+ boolean meetException = false;
+ String dbName = spec.getTable().getDbName();
+ String tableName = spec.getTable().getTableName();
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ for (MetadataWriter writer : metadataWriters) {
+ if (meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ } catch (Exception e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ }
+ private void addException(Exception e, String tableString, String dbName, String tableName) throws IOException{
+ TableStatus tableStatus = tableOperationTypeMap.get(tableString);
+ Map<String, Queue<MetadataFlushException>> tableErrorMap = this.datasetErrorMap.getOrDefault(tableStatus.datasetName, new HashMap<>());
+ Queue<MetadataFlushException> errors = tableErrorMap.getOrDefault(tableString, new LinkedList<>());
+ if (!errors.isEmpty() && errors.peek().highWatermark == tableStatus.gmceLowWatermark) {
+ errors.peek().highWatermark = tableStatus.gmceHighWatermark;
+ } else {
+ MetadataFlushException metadataFlushException =
+ new MetadataFlushException(tableStatus.datasetName, dbName, tableName, tableStatus.gmceTopicName, tableStatus.gmceTopicPartition, tableStatus.gmceLowWatermark, tableStatus.gmceHighWatermark, e);
+ errors.offer(metadataFlushException);
+ }
+ tableErrorMap.put(tableString, errors);
+ this.datasetErrorMap.put(tableStatus.datasetName, tableErrorMap);
Review comment:
Sorry for that confusing, I was using datasetName to be consistent with the metadata schema where they treat path as datasetName. I will change datasetName to datasetPath to avoid this confusion. It's not a 1:1 mapping, as one dataset can be register into multi tables
--
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 edited a comment on pull request #3419: [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #3419:
URL: https://github.com/apache/gobblin/pull/3419#issuecomment-951356726
# [Codecov](https://codecov.io/gh/apache/gobblin/pull/3419?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 [#3419](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (c6bcb1c) into [master](https://codecov.io/gh/apache/gobblin/commit/255bdc15f4e8dd7dbf5fb94e9742d8430f2506f3?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (255bdc1) will **increase** coverage by `1.89%`.
> The diff coverage is `73.80%`.
[](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #3419 +/- ##
============================================
+ Coverage 46.53% 48.42% +1.89%
+ Complexity 10278 7683 -2595
============================================
Files 2065 1437 -628
Lines 80638 56774 -23864
Branches 9005 6542 -2463
============================================
- Hits 37521 27493 -10028
+ Misses 39641 26711 -12930
+ Partials 3476 2570 -906
```
| [Impacted Files](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...apache/gobblin/hive/writer/HiveMetadataWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1oaXZlLXJlZ2lzdHJhdGlvbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZ29iYmxpbi9oaXZlL3dyaXRlci9IaXZlTWV0YWRhdGFXcml0ZXIuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...pache/gobblin/iceberg/writer/GobblinMCEWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvd3JpdGVyL0dvYmJsaW5NQ0VXcml0ZXIuamF2YQ==) | `74.01% <75.75%> (+1.65%)` | :arrow_up: |
| [...bblin/iceberg/writer/GobblinMetadataException.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvd3JpdGVyL0dvYmJsaW5NZXRhZGF0YUV4Y2VwdGlvbi5qYXZh) | `100.00% <100.00%> (ø)` | |
| [.../gobblin/iceberg/writer/IcebergMetadataWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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) | `72.21% <100.00%> (+0.10%)` | :arrow_up: |
| [.../java/org/apache/gobblin/cluster/SleepingTask.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1jbHVzdGVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NsdXN0ZXIvU2xlZXBpbmdUYXNrLmphdmE=) | `39.39% <0.00%> (-6.07%)` | :arrow_down: |
| [...main/java/org/apache/gobblin/yarn/YarnService.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi15YXJuL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3lhcm4vWWFyblNlcnZpY2UuamF2YQ==) | `15.46% <0.00%> (-0.78%)` | :arrow_down: |
| [...apache/gobblin/salesforce/SalesforceExtractor.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1zYWxlc2ZvcmNlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3NhbGVzZm9yY2UvU2FsZXNmb3JjZUV4dHJhY3Rvci5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...lin/runtime/job\_catalog/ImmutableFSJobCatalog.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3J1bnRpbWUvam9iX2NhdGFsb2cvSW1tdXRhYmxlRlNKb2JDYXRhbG9nLmphdmE=) | `75.22% <0.00%> (ø)` | |
| [.../profile/ManagedIcebergCleanableDatasetFinder.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1kYXRhLW1hbmFnZW1lbnQvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2dvYmJsaW4vZGF0YS9tYW5hZ2VtZW50L3JldGVudGlvbi9wcm9maWxlL01hbmFnZWRJY2ViZXJnQ2xlYW5hYmxlRGF0YXNldEZpbmRlci5qYXZh) | `0.00% <0.00%> (ø)` | |
| ... and [644 more](https://codecov.io/gh/apache/gobblin/pull/3419/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/3419?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/3419?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 [255bdc1...c6bcb1c](https://codecov.io/gh/apache/gobblin/pull/3419?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] codecov-commenter edited a comment on pull request #3419: [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #3419:
URL: https://github.com/apache/gobblin/pull/3419#issuecomment-951356726
# [Codecov](https://codecov.io/gh/apache/gobblin/pull/3419?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 [#3419](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (1a8a00a) into [master](https://codecov.io/gh/apache/gobblin/commit/255bdc15f4e8dd7dbf5fb94e9742d8430f2506f3?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (255bdc1) will **increase** coverage by `1.85%`.
> The diff coverage is `71.25%`.
[](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #3419 +/- ##
============================================
+ Coverage 46.53% 48.38% +1.85%
+ Complexity 10278 7658 -2620
============================================
Files 2065 1432 -633
Lines 80638 56659 -23979
Branches 9005 6538 -2467
============================================
- Hits 37521 27416 -10105
+ Misses 39641 26675 -12966
+ Partials 3476 2568 -908
```
| [Impacted Files](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...apache/gobblin/hive/writer/HiveMetadataWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1oaXZlLXJlZ2lzdHJhdGlvbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZ29iYmxpbi9oaXZlL3dyaXRlci9IaXZlTWV0YWRhdGFXcml0ZXIuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [.../gobblin/iceberg/writer/IcebergMetadataWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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) | `71.84% <0.00%> (-0.27%)` | :arrow_down: |
| [...pache/gobblin/iceberg/writer/GobblinMCEWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvd3JpdGVyL0dvYmJsaW5NQ0VXcml0ZXIuamF2YQ==) | `74.13% <76.19%> (+1.78%)` | :arrow_up: |
| [...gobblin/iceberg/writer/MetadataFlushException.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvd3JpdGVyL01ldGFkYXRhRmx1c2hFeGNlcHRpb24uamF2YQ==) | `100.00% <100.00%> (ø)` | |
| [.../java/org/apache/gobblin/cluster/SleepingTask.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1jbHVzdGVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NsdXN0ZXIvU2xlZXBpbmdUYXNrLmphdmE=) | `39.39% <0.00%> (-6.07%)` | :arrow_down: |
| [...a/management/copy/publisher/CopyDataPublisher.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1kYXRhLW1hbmFnZW1lbnQvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2dvYmJsaW4vZGF0YS9tYW5hZ2VtZW50L2NvcHkvcHVibGlzaGVyL0NvcHlEYXRhUHVibGlzaGVyLmphdmE=) | `74.17% <0.00%> (-1.33%)` | :arrow_down: |
| [.../apache/gobblin/runtime/api/JobExecutionState.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3J1bnRpbWUvYXBpL0pvYkV4ZWN1dGlvblN0YXRlLmphdmE=) | `79.43% <0.00%> (-0.94%)` | :arrow_down: |
| [...main/java/org/apache/gobblin/yarn/YarnService.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi15YXJuL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3lhcm4vWWFyblNlcnZpY2UuamF2YQ==) | `15.46% <0.00%> (-0.78%)` | :arrow_down: |
| [...main/java/org/apache/gobblin/util/HadoopUtils.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi11dGlsaXR5L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3V0aWwvSGFkb29wVXRpbHMuamF2YQ==) | `29.93% <0.00%> (-0.66%)` | :arrow_down: |
| ... and [637 more](https://codecov.io/gh/apache/gobblin/pull/3419/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/3419?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/3419?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 [255bdc1...1a8a00a](https://codecov.io/gh/apache/gobblin/pull/3419?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] codecov-commenter edited a comment on pull request #3419: [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #3419:
URL: https://github.com/apache/gobblin/pull/3419#issuecomment-951356726
# [Codecov](https://codecov.io/gh/apache/gobblin/pull/3419?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 [#3419](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (1a8a00a) into [master](https://codecov.io/gh/apache/gobblin/commit/255bdc15f4e8dd7dbf5fb94e9742d8430f2506f3?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (255bdc1) will **increase** coverage by `2.55%`.
> The diff coverage is `71.25%`.
[](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #3419 +/- ##
============================================
+ Coverage 46.53% 49.08% +2.55%
+ Complexity 10278 8824 -1454
============================================
Files 2065 1685 -380
Lines 80638 65165 -15473
Branches 9005 7478 -1527
============================================
- Hits 37521 31985 -5536
+ Misses 39641 30153 -9488
+ Partials 3476 3027 -449
```
| [Impacted Files](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...apache/gobblin/hive/writer/HiveMetadataWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1oaXZlLXJlZ2lzdHJhdGlvbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZ29iYmxpbi9oaXZlL3dyaXRlci9IaXZlTWV0YWRhdGFXcml0ZXIuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [.../gobblin/iceberg/writer/IcebergMetadataWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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) | `71.84% <0.00%> (-0.27%)` | :arrow_down: |
| [...pache/gobblin/iceberg/writer/GobblinMCEWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvd3JpdGVyL0dvYmJsaW5NQ0VXcml0ZXIuamF2YQ==) | `74.13% <76.19%> (+1.78%)` | :arrow_up: |
| [...gobblin/iceberg/writer/MetadataFlushException.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvd3JpdGVyL01ldGFkYXRhRmx1c2hFeGNlcHRpb24uamF2YQ==) | `100.00% <100.00%> (ø)` | |
| [.../java/org/apache/gobblin/cluster/SleepingTask.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1jbHVzdGVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NsdXN0ZXIvU2xlZXBpbmdUYXNrLmphdmE=) | `39.39% <0.00%> (-6.07%)` | :arrow_down: |
| [...lin/restli/throttling/ZookeeperLeaderElection.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1yZXN0bGkvZ29iYmxpbi10aHJvdHRsaW5nLXNlcnZpY2UvZ29iYmxpbi10aHJvdHRsaW5nLXNlcnZpY2Utc2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3Jlc3RsaS90aHJvdHRsaW5nL1pvb2tlZXBlckxlYWRlckVsZWN0aW9uLmphdmE=) | `70.00% <0.00%> (-2.23%)` | :arrow_down: |
| [...a/management/copy/publisher/CopyDataPublisher.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1kYXRhLW1hbmFnZW1lbnQvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2dvYmJsaW4vZGF0YS9tYW5hZ2VtZW50L2NvcHkvcHVibGlzaGVyL0NvcHlEYXRhUHVibGlzaGVyLmphdmE=) | `74.17% <0.00%> (-1.33%)` | :arrow_down: |
| [.../apache/gobblin/runtime/api/JobExecutionState.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3J1bnRpbWUvYXBpL0pvYkV4ZWN1dGlvblN0YXRlLmphdmE=) | `79.43% <0.00%> (-0.94%)` | :arrow_down: |
| [...main/java/org/apache/gobblin/yarn/YarnService.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi15YXJuL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3lhcm4vWWFyblNlcnZpY2UuamF2YQ==) | `15.46% <0.00%> (-0.78%)` | :arrow_down: |
| [...main/java/org/apache/gobblin/util/HadoopUtils.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi11dGlsaXR5L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3V0aWwvSGFkb29wVXRpbHMuamF2YQ==) | `29.93% <0.00%> (-0.66%)` | :arrow_down: |
| ... and [385 more](https://codecov.io/gh/apache/gobblin/pull/3419/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/3419?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/3419?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 [255bdc1...1a8a00a](https://codecov.io/gh/apache/gobblin/pull/3419?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 a change in pull request #3419: [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container
Posted by GitBox <gi...@apache.org>.
sv2000 commented on a change in pull request #3419:
URL: https://github.com/apache/gobblin/pull/3419#discussion_r740443463
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -248,20 +272,88 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
String dbName = spec.getTable().getDbName();
String tableName = spec.getTable().getTableName();
String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
- if (tableOperationTypeMap.containsKey(tableString)
- && tableOperationTypeMap.get(tableString) != gmce.getOperationType()) {
- for (MetadataWriter writer : metadataWriters) {
- writer.flush(dbName, tableName);
- }
- }
- tableOperationTypeMap.put(tableString, gmce.getOperationType());
- for (MetadataWriter writer : metadataWriters) {
- writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ if (!tableOperationTypeMap.containsKey(tableString)) {
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
+ } else if (tableOperationTypeMap.get(tableString).operationType != gmce.getOperationType()) {
+ flush(dbName, tableName);
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
}
+ tableOperationTypeMap.get(tableString).gmceHighWatermark = watermark.getLwm().getValue();
+ write(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
}
this.recordCount.incrementAndGet();
}
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void write(RecordEnvelope recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
+ boolean meetException = false;
+ String dbName = spec.getTable().getDbName();
+ String tableName = spec.getTable().getTableName();
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ for (MetadataWriter writer : metadataWriters) {
+ if (meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ } catch (Exception e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ }
+ private void addException(Exception e, String tableString, String dbName, String tableName) throws IOException{
Review comment:
Add new line before the start of a new method.
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -248,20 +272,88 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
String dbName = spec.getTable().getDbName();
String tableName = spec.getTable().getTableName();
String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
- if (tableOperationTypeMap.containsKey(tableString)
- && tableOperationTypeMap.get(tableString) != gmce.getOperationType()) {
- for (MetadataWriter writer : metadataWriters) {
- writer.flush(dbName, tableName);
- }
- }
- tableOperationTypeMap.put(tableString, gmce.getOperationType());
- for (MetadataWriter writer : metadataWriters) {
- writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ if (!tableOperationTypeMap.containsKey(tableString)) {
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
+ } else if (tableOperationTypeMap.get(tableString).operationType != gmce.getOperationType()) {
+ flush(dbName, tableName);
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
}
+ tableOperationTypeMap.get(tableString).gmceHighWatermark = watermark.getLwm().getValue();
+ write(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
}
this.recordCount.incrementAndGet();
}
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void write(RecordEnvelope recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
+ boolean meetException = false;
+ String dbName = spec.getTable().getDbName();
+ String tableName = spec.getTable().getTableName();
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ for (MetadataWriter writer : metadataWriters) {
+ if (meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ } catch (Exception e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ }
+ private void addException(Exception e, String tableString, String dbName, String tableName) throws IOException{
+ TableStatus tableStatus = tableOperationTypeMap.get(tableString);
+ Map<String, Queue<MetadataFlushException>> tableErrorMap = this.datasetErrorMap.getOrDefault(tableStatus.datasetName, new HashMap<>());
+ Queue<MetadataFlushException> errors = tableErrorMap.getOrDefault(tableString, new LinkedList<>());
+ if (!errors.isEmpty() && errors.peek().highWatermark == tableStatus.gmceLowWatermark) {
+ errors.peek().highWatermark = tableStatus.gmceHighWatermark;
+ } else {
+ MetadataFlushException metadataFlushException =
+ new MetadataFlushException(tableStatus.datasetName, dbName, tableName, tableStatus.gmceTopicName, tableStatus.gmceTopicPartition, tableStatus.gmceLowWatermark, tableStatus.gmceHighWatermark, e);
+ errors.offer(metadataFlushException);
+ }
+ tableErrorMap.put(tableString, errors);
+ this.datasetErrorMap.put(tableStatus.datasetName, tableErrorMap);
Review comment:
This is confusing. How is datasetName different from tableString? If there is a 1:1 mapping between datasetName and tableString, is Map of Maps the right data structure here?
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -187,6 +210,7 @@ public Descriptor getDataDescriptor() {
@Override
public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws IOException {
GenericRecord genericRecord = recordEnvelope.getRecord();
+ KafkaStreamingExtractor.KafkaWatermark watermark = ((KafkaStreamingExtractor.KafkaWatermark) recordEnvelope.getWatermark());
Review comment:
Looks like we are leaking details here. Ideally, we should not expose details of extractor implementation here. Can we use the CheckpointableWatermark interface instead?
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/MetadataFlushException.java
##########
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.iceberg.writer;
+
+import java.io.IOException;
+
+
+public class MetadataFlushException extends IOException {
+ public String datasetName;
+ public String dbName;
+ public String tableName;
+ public String GMCETopicName;
+ public int GMCETopicPartition;
+ public long highWatermark;
+ public long lowWatermark;
+ public Exception exception;
+ MetadataFlushException(String datasetName, String dbName, String tableName, String GMCETopicName,
Review comment:
Since this is used in both write and flush, shall we call it GMCEException or something more generic?
##########
File path: gobblin-hive-registration/src/main/java/org/apache/gobblin/hive/writer/MetadataWriter.java
##########
@@ -47,6 +47,16 @@
*/
void flush(String dbName, String tableName) throws IOException;
+ /**
+ * If something wrong happens, we want to reset the table inside the writer so that we can continual
Review comment:
continual -> continue
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -248,20 +272,88 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
String dbName = spec.getTable().getDbName();
String tableName = spec.getTable().getTableName();
String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
- if (tableOperationTypeMap.containsKey(tableString)
- && tableOperationTypeMap.get(tableString) != gmce.getOperationType()) {
- for (MetadataWriter writer : metadataWriters) {
- writer.flush(dbName, tableName);
- }
- }
- tableOperationTypeMap.put(tableString, gmce.getOperationType());
- for (MetadataWriter writer : metadataWriters) {
- writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ if (!tableOperationTypeMap.containsKey(tableString)) {
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
+ } else if (tableOperationTypeMap.get(tableString).operationType != gmce.getOperationType()) {
+ flush(dbName, tableName);
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
}
+ tableOperationTypeMap.get(tableString).gmceHighWatermark = watermark.getLwm().getValue();
+ write(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
}
this.recordCount.incrementAndGet();
}
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void write(RecordEnvelope recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
+ boolean meetException = false;
+ String dbName = spec.getTable().getDbName();
+ String tableName = spec.getTable().getTableName();
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ for (MetadataWriter writer : metadataWriters) {
+ if (meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ } catch (Exception e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ }
+ private void addException(Exception e, String tableString, String dbName, String tableName) throws IOException{
+ TableStatus tableStatus = tableOperationTypeMap.get(tableString);
+ Map<String, Queue<MetadataFlushException>> tableErrorMap = this.datasetErrorMap.getOrDefault(tableStatus.datasetName, new HashMap<>());
+ Queue<MetadataFlushException> errors = tableErrorMap.getOrDefault(tableString, new LinkedList<>());
+ if (!errors.isEmpty() && errors.peek().highWatermark == tableStatus.gmceLowWatermark) {
+ errors.peek().highWatermark = tableStatus.gmceHighWatermark;
+ } else {
+ MetadataFlushException metadataFlushException =
+ new MetadataFlushException(tableStatus.datasetName, dbName, tableName, tableStatus.gmceTopicName, tableStatus.gmceTopicPartition, tableStatus.gmceLowWatermark, tableStatus.gmceHighWatermark, e);
+ errors.offer(metadataFlushException);
+ }
+ tableErrorMap.put(tableString, errors);
+ this.datasetErrorMap.put(tableStatus.datasetName, tableErrorMap);
+ log.error(String.format("Meet exception when flush table %s", tableString), e);
+ if (datasetErrorMap.size() > maxErrorDataset) {
+ //Fail the job if the error size exceeds some number
+ throw new IOException(String.format("Container fails to flush for more than %s dataset", maxErrorDataset));
+ }
+ tableStatus.gmceLowWatermark = tableStatus.gmceHighWatermark;
+ }
+
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void flush(String dbName, String tableName) throws IOException {
+ boolean meetException = false;
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ if (tableOperationTypeMap.get(tableString).gmceLowWatermark == tableOperationTypeMap.get(tableString).gmceHighWatermark) {
+ // No need to flush
+ return;
+ }
+ for (MetadataWriter writer : metadataWriters) {
+ if(meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.flush(dbName, tableName);
+ } catch (IOException e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ if (!meetException && datasetErrorMap.containsKey(tableOperationTypeMap.get(tableString).datasetName)
Review comment:
Shouldn't this if condition be triggered if meetException = true?
##########
File path: gobblin-hive-registration/src/main/java/org/apache/gobblin/hive/writer/MetadataWriter.java
##########
@@ -47,6 +47,16 @@
*/
void flush(String dbName, String tableName) throws IOException;
+ /**
+ * If something wrong happens, we want to reset the table inside the writer so that we can continual
Review comment:
The meaning of reset is not quite clear from the javadoc description. Is it about cleaning up some in-memory state that is maintained by the pipeline for the table?
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -248,20 +272,88 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
String dbName = spec.getTable().getDbName();
String tableName = spec.getTable().getTableName();
String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
- if (tableOperationTypeMap.containsKey(tableString)
- && tableOperationTypeMap.get(tableString) != gmce.getOperationType()) {
- for (MetadataWriter writer : metadataWriters) {
- writer.flush(dbName, tableName);
- }
- }
- tableOperationTypeMap.put(tableString, gmce.getOperationType());
- for (MetadataWriter writer : metadataWriters) {
- writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ if (!tableOperationTypeMap.containsKey(tableString)) {
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
+ } else if (tableOperationTypeMap.get(tableString).operationType != gmce.getOperationType()) {
+ flush(dbName, tableName);
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
}
+ tableOperationTypeMap.get(tableString).gmceHighWatermark = watermark.getLwm().getValue();
+ write(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
}
this.recordCount.incrementAndGet();
}
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void write(RecordEnvelope recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
+ boolean meetException = false;
+ String dbName = spec.getTable().getDbName();
+ String tableName = spec.getTable().getTableName();
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ for (MetadataWriter writer : metadataWriters) {
+ if (meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ } catch (Exception e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ }
+ private void addException(Exception e, String tableString, String dbName, String tableName) throws IOException{
+ TableStatus tableStatus = tableOperationTypeMap.get(tableString);
+ Map<String, Queue<MetadataFlushException>> tableErrorMap = this.datasetErrorMap.getOrDefault(tableStatus.datasetName, new HashMap<>());
+ Queue<MetadataFlushException> errors = tableErrorMap.getOrDefault(tableString, new LinkedList<>());
+ if (!errors.isEmpty() && errors.peek().highWatermark == tableStatus.gmceLowWatermark) {
+ errors.peek().highWatermark = tableStatus.gmceHighWatermark;
+ } else {
+ MetadataFlushException metadataFlushException =
+ new MetadataFlushException(tableStatus.datasetName, dbName, tableName, tableStatus.gmceTopicName, tableStatus.gmceTopicPartition, tableStatus.gmceLowWatermark, tableStatus.gmceHighWatermark, e);
+ errors.offer(metadataFlushException);
+ }
+ tableErrorMap.put(tableString, errors);
+ this.datasetErrorMap.put(tableStatus.datasetName, tableErrorMap);
+ log.error(String.format("Meet exception when flush table %s", tableString), e);
+ if (datasetErrorMap.size() > maxErrorDataset) {
+ //Fail the job if the error size exceeds some number
+ throw new IOException(String.format("Container fails to flush for more than %s dataset", maxErrorDataset));
Review comment:
Should we throw the actual exception that the flush/write encountered for the specific dataset?
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -248,20 +272,88 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
String dbName = spec.getTable().getDbName();
String tableName = spec.getTable().getTableName();
String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
- if (tableOperationTypeMap.containsKey(tableString)
- && tableOperationTypeMap.get(tableString) != gmce.getOperationType()) {
- for (MetadataWriter writer : metadataWriters) {
- writer.flush(dbName, tableName);
- }
- }
- tableOperationTypeMap.put(tableString, gmce.getOperationType());
- for (MetadataWriter writer : metadataWriters) {
- writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ if (!tableOperationTypeMap.containsKey(tableString)) {
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
+ } else if (tableOperationTypeMap.get(tableString).operationType != gmce.getOperationType()) {
+ flush(dbName, tableName);
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
}
+ tableOperationTypeMap.get(tableString).gmceHighWatermark = watermark.getLwm().getValue();
+ write(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
}
this.recordCount.incrementAndGet();
}
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void write(RecordEnvelope recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
+ boolean meetException = false;
+ String dbName = spec.getTable().getDbName();
+ String tableName = spec.getTable().getTableName();
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ for (MetadataWriter writer : metadataWriters) {
+ if (meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ } catch (Exception e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ }
+ private void addException(Exception e, String tableString, String dbName, String tableName) throws IOException{
Review comment:
Should we call this method addOrThrowException?
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/MetadataFlushException.java
##########
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.iceberg.writer;
+
+import java.io.IOException;
+
+
+public class MetadataFlushException extends IOException {
+ public String datasetName;
Review comment:
Seems like there is a lot of overlap in the fields: datasetName, dbName, tableName, topicName, topicPartition. Can we keep only the ones needed and derive the rest from those fields?
--
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 change in pull request #3419: [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container
Posted by GitBox <gi...@apache.org>.
sv2000 commented on a change in pull request #3419:
URL: https://github.com/apache/gobblin/pull/3419#discussion_r740443463
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -248,20 +272,88 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
String dbName = spec.getTable().getDbName();
String tableName = spec.getTable().getTableName();
String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
- if (tableOperationTypeMap.containsKey(tableString)
- && tableOperationTypeMap.get(tableString) != gmce.getOperationType()) {
- for (MetadataWriter writer : metadataWriters) {
- writer.flush(dbName, tableName);
- }
- }
- tableOperationTypeMap.put(tableString, gmce.getOperationType());
- for (MetadataWriter writer : metadataWriters) {
- writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ if (!tableOperationTypeMap.containsKey(tableString)) {
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
+ } else if (tableOperationTypeMap.get(tableString).operationType != gmce.getOperationType()) {
+ flush(dbName, tableName);
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
}
+ tableOperationTypeMap.get(tableString).gmceHighWatermark = watermark.getLwm().getValue();
+ write(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
}
this.recordCount.incrementAndGet();
}
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void write(RecordEnvelope recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
+ boolean meetException = false;
+ String dbName = spec.getTable().getDbName();
+ String tableName = spec.getTable().getTableName();
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ for (MetadataWriter writer : metadataWriters) {
+ if (meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ } catch (Exception e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ }
+ private void addException(Exception e, String tableString, String dbName, String tableName) throws IOException{
Review comment:
Add new line before the start of a new method.
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -248,20 +272,88 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
String dbName = spec.getTable().getDbName();
String tableName = spec.getTable().getTableName();
String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
- if (tableOperationTypeMap.containsKey(tableString)
- && tableOperationTypeMap.get(tableString) != gmce.getOperationType()) {
- for (MetadataWriter writer : metadataWriters) {
- writer.flush(dbName, tableName);
- }
- }
- tableOperationTypeMap.put(tableString, gmce.getOperationType());
- for (MetadataWriter writer : metadataWriters) {
- writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ if (!tableOperationTypeMap.containsKey(tableString)) {
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
+ } else if (tableOperationTypeMap.get(tableString).operationType != gmce.getOperationType()) {
+ flush(dbName, tableName);
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
}
+ tableOperationTypeMap.get(tableString).gmceHighWatermark = watermark.getLwm().getValue();
+ write(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
}
this.recordCount.incrementAndGet();
}
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void write(RecordEnvelope recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
+ boolean meetException = false;
+ String dbName = spec.getTable().getDbName();
+ String tableName = spec.getTable().getTableName();
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ for (MetadataWriter writer : metadataWriters) {
+ if (meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ } catch (Exception e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ }
+ private void addException(Exception e, String tableString, String dbName, String tableName) throws IOException{
+ TableStatus tableStatus = tableOperationTypeMap.get(tableString);
+ Map<String, Queue<MetadataFlushException>> tableErrorMap = this.datasetErrorMap.getOrDefault(tableStatus.datasetName, new HashMap<>());
+ Queue<MetadataFlushException> errors = tableErrorMap.getOrDefault(tableString, new LinkedList<>());
+ if (!errors.isEmpty() && errors.peek().highWatermark == tableStatus.gmceLowWatermark) {
+ errors.peek().highWatermark = tableStatus.gmceHighWatermark;
+ } else {
+ MetadataFlushException metadataFlushException =
+ new MetadataFlushException(tableStatus.datasetName, dbName, tableName, tableStatus.gmceTopicName, tableStatus.gmceTopicPartition, tableStatus.gmceLowWatermark, tableStatus.gmceHighWatermark, e);
+ errors.offer(metadataFlushException);
+ }
+ tableErrorMap.put(tableString, errors);
+ this.datasetErrorMap.put(tableStatus.datasetName, tableErrorMap);
Review comment:
This is confusing. How is datasetName different from tableString? If there is a 1:1 mapping between datasetName and tableString, is Map of Maps the right data structure here?
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -187,6 +210,7 @@ public Descriptor getDataDescriptor() {
@Override
public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws IOException {
GenericRecord genericRecord = recordEnvelope.getRecord();
+ KafkaStreamingExtractor.KafkaWatermark watermark = ((KafkaStreamingExtractor.KafkaWatermark) recordEnvelope.getWatermark());
Review comment:
Looks like we are leaking details here. Ideally, we should not expose details of extractor implementation here. Can we use the CheckpointableWatermark interface instead?
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/MetadataFlushException.java
##########
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.iceberg.writer;
+
+import java.io.IOException;
+
+
+public class MetadataFlushException extends IOException {
+ public String datasetName;
+ public String dbName;
+ public String tableName;
+ public String GMCETopicName;
+ public int GMCETopicPartition;
+ public long highWatermark;
+ public long lowWatermark;
+ public Exception exception;
+ MetadataFlushException(String datasetName, String dbName, String tableName, String GMCETopicName,
Review comment:
Since this is used in both write and flush, shall we call it GMCEException or something more generic?
##########
File path: gobblin-hive-registration/src/main/java/org/apache/gobblin/hive/writer/MetadataWriter.java
##########
@@ -47,6 +47,16 @@
*/
void flush(String dbName, String tableName) throws IOException;
+ /**
+ * If something wrong happens, we want to reset the table inside the writer so that we can continual
Review comment:
continual -> continue
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -248,20 +272,88 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
String dbName = spec.getTable().getDbName();
String tableName = spec.getTable().getTableName();
String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
- if (tableOperationTypeMap.containsKey(tableString)
- && tableOperationTypeMap.get(tableString) != gmce.getOperationType()) {
- for (MetadataWriter writer : metadataWriters) {
- writer.flush(dbName, tableName);
- }
- }
- tableOperationTypeMap.put(tableString, gmce.getOperationType());
- for (MetadataWriter writer : metadataWriters) {
- writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ if (!tableOperationTypeMap.containsKey(tableString)) {
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
+ } else if (tableOperationTypeMap.get(tableString).operationType != gmce.getOperationType()) {
+ flush(dbName, tableName);
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
}
+ tableOperationTypeMap.get(tableString).gmceHighWatermark = watermark.getLwm().getValue();
+ write(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
}
this.recordCount.incrementAndGet();
}
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void write(RecordEnvelope recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
+ boolean meetException = false;
+ String dbName = spec.getTable().getDbName();
+ String tableName = spec.getTable().getTableName();
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ for (MetadataWriter writer : metadataWriters) {
+ if (meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ } catch (Exception e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ }
+ private void addException(Exception e, String tableString, String dbName, String tableName) throws IOException{
+ TableStatus tableStatus = tableOperationTypeMap.get(tableString);
+ Map<String, Queue<MetadataFlushException>> tableErrorMap = this.datasetErrorMap.getOrDefault(tableStatus.datasetName, new HashMap<>());
+ Queue<MetadataFlushException> errors = tableErrorMap.getOrDefault(tableString, new LinkedList<>());
+ if (!errors.isEmpty() && errors.peek().highWatermark == tableStatus.gmceLowWatermark) {
+ errors.peek().highWatermark = tableStatus.gmceHighWatermark;
+ } else {
+ MetadataFlushException metadataFlushException =
+ new MetadataFlushException(tableStatus.datasetName, dbName, tableName, tableStatus.gmceTopicName, tableStatus.gmceTopicPartition, tableStatus.gmceLowWatermark, tableStatus.gmceHighWatermark, e);
+ errors.offer(metadataFlushException);
+ }
+ tableErrorMap.put(tableString, errors);
+ this.datasetErrorMap.put(tableStatus.datasetName, tableErrorMap);
+ log.error(String.format("Meet exception when flush table %s", tableString), e);
+ if (datasetErrorMap.size() > maxErrorDataset) {
+ //Fail the job if the error size exceeds some number
+ throw new IOException(String.format("Container fails to flush for more than %s dataset", maxErrorDataset));
+ }
+ tableStatus.gmceLowWatermark = tableStatus.gmceHighWatermark;
+ }
+
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void flush(String dbName, String tableName) throws IOException {
+ boolean meetException = false;
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ if (tableOperationTypeMap.get(tableString).gmceLowWatermark == tableOperationTypeMap.get(tableString).gmceHighWatermark) {
+ // No need to flush
+ return;
+ }
+ for (MetadataWriter writer : metadataWriters) {
+ if(meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.flush(dbName, tableName);
+ } catch (IOException e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ if (!meetException && datasetErrorMap.containsKey(tableOperationTypeMap.get(tableString).datasetName)
Review comment:
Shouldn't this if condition be triggered if meetException = true?
##########
File path: gobblin-hive-registration/src/main/java/org/apache/gobblin/hive/writer/MetadataWriter.java
##########
@@ -47,6 +47,16 @@
*/
void flush(String dbName, String tableName) throws IOException;
+ /**
+ * If something wrong happens, we want to reset the table inside the writer so that we can continual
Review comment:
The meaning of reset is not quite clear from the javadoc description. Is it about cleaning up some in-memory state that is maintained by the pipeline for the table?
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -248,20 +272,88 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
String dbName = spec.getTable().getDbName();
String tableName = spec.getTable().getTableName();
String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
- if (tableOperationTypeMap.containsKey(tableString)
- && tableOperationTypeMap.get(tableString) != gmce.getOperationType()) {
- for (MetadataWriter writer : metadataWriters) {
- writer.flush(dbName, tableName);
- }
- }
- tableOperationTypeMap.put(tableString, gmce.getOperationType());
- for (MetadataWriter writer : metadataWriters) {
- writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ if (!tableOperationTypeMap.containsKey(tableString)) {
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
+ } else if (tableOperationTypeMap.get(tableString).operationType != gmce.getOperationType()) {
+ flush(dbName, tableName);
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
}
+ tableOperationTypeMap.get(tableString).gmceHighWatermark = watermark.getLwm().getValue();
+ write(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
}
this.recordCount.incrementAndGet();
}
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void write(RecordEnvelope recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
+ boolean meetException = false;
+ String dbName = spec.getTable().getDbName();
+ String tableName = spec.getTable().getTableName();
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ for (MetadataWriter writer : metadataWriters) {
+ if (meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ } catch (Exception e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ }
+ private void addException(Exception e, String tableString, String dbName, String tableName) throws IOException{
+ TableStatus tableStatus = tableOperationTypeMap.get(tableString);
+ Map<String, Queue<MetadataFlushException>> tableErrorMap = this.datasetErrorMap.getOrDefault(tableStatus.datasetName, new HashMap<>());
+ Queue<MetadataFlushException> errors = tableErrorMap.getOrDefault(tableString, new LinkedList<>());
+ if (!errors.isEmpty() && errors.peek().highWatermark == tableStatus.gmceLowWatermark) {
+ errors.peek().highWatermark = tableStatus.gmceHighWatermark;
+ } else {
+ MetadataFlushException metadataFlushException =
+ new MetadataFlushException(tableStatus.datasetName, dbName, tableName, tableStatus.gmceTopicName, tableStatus.gmceTopicPartition, tableStatus.gmceLowWatermark, tableStatus.gmceHighWatermark, e);
+ errors.offer(metadataFlushException);
+ }
+ tableErrorMap.put(tableString, errors);
+ this.datasetErrorMap.put(tableStatus.datasetName, tableErrorMap);
+ log.error(String.format("Meet exception when flush table %s", tableString), e);
+ if (datasetErrorMap.size() > maxErrorDataset) {
+ //Fail the job if the error size exceeds some number
+ throw new IOException(String.format("Container fails to flush for more than %s dataset", maxErrorDataset));
Review comment:
Should we throw the actual exception that the flush/write encountered for the specific dataset?
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -248,20 +272,88 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
String dbName = spec.getTable().getDbName();
String tableName = spec.getTable().getTableName();
String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
- if (tableOperationTypeMap.containsKey(tableString)
- && tableOperationTypeMap.get(tableString) != gmce.getOperationType()) {
- for (MetadataWriter writer : metadataWriters) {
- writer.flush(dbName, tableName);
- }
- }
- tableOperationTypeMap.put(tableString, gmce.getOperationType());
- for (MetadataWriter writer : metadataWriters) {
- writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ if (!tableOperationTypeMap.containsKey(tableString)) {
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
+ } else if (tableOperationTypeMap.get(tableString).operationType != gmce.getOperationType()) {
+ flush(dbName, tableName);
+ tableOperationTypeMap.put(tableString, new TableStatus(gmce.getOperationType(),
+ gmce.getDatasetIdentifier().getNativeName(), watermark.getTopicPartition().getTopicName(), watermark.getTopicPartition().getId(),
+ watermark.getLwm().getValue()-1, watermark.getLwm().getValue()));
}
+ tableOperationTypeMap.get(tableString).gmceHighWatermark = watermark.getLwm().getValue();
+ write(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
}
this.recordCount.incrementAndGet();
}
+ // Add fault tolerant ability and make sure we can emit GTE as desired
+ private void write(RecordEnvelope recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
+ boolean meetException = false;
+ String dbName = spec.getTable().getDbName();
+ String tableName = spec.getTable().getTableName();
+ String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
+ for (MetadataWriter writer : metadataWriters) {
+ if (meetException) {
+ writer.reset(dbName, tableName);
+ } else {
+ try {
+ writer.writeEnvelope(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+ } catch (Exception e) {
+ meetException = true;
+ addException(e, tableString, dbName, tableName);
+ }
+ }
+ }
+ }
+ private void addException(Exception e, String tableString, String dbName, String tableName) throws IOException{
Review comment:
Should we call this method addOrThrowException?
##########
File path: gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/MetadataFlushException.java
##########
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.iceberg.writer;
+
+import java.io.IOException;
+
+
+public class MetadataFlushException extends IOException {
+ public String datasetName;
Review comment:
Seems like there is a lot of overlap in the fields: datasetName, dbName, tableName, topicName, topicPartition. Can we keep only the ones needed and derive the rest from those fields?
--
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 edited a comment on pull request #3419: [GOBBLIN-1565]Make GMCEWriter fault tolerant so that one topic failure will not affect other topics in the same container
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #3419:
URL: https://github.com/apache/gobblin/pull/3419#issuecomment-951356726
# [Codecov](https://codecov.io/gh/apache/gobblin/pull/3419?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 [#3419](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (c6bcb1c) into [master](https://codecov.io/gh/apache/gobblin/commit/255bdc15f4e8dd7dbf5fb94e9742d8430f2506f3?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (255bdc1) will **increase** coverage by `0.04%`.
> The diff coverage is `73.80%`.
[](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #3419 +/- ##
============================================
+ Coverage 46.53% 46.57% +0.04%
- Complexity 10278 10313 +35
============================================
Files 2065 2071 +6
Lines 80638 80821 +183
Branches 9005 9015 +10
============================================
+ Hits 37521 37641 +120
- Misses 39641 39698 +57
- Partials 3476 3482 +6
```
| [Impacted Files](https://codecov.io/gh/apache/gobblin/pull/3419?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...apache/gobblin/hive/writer/HiveMetadataWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1oaXZlLXJlZ2lzdHJhdGlvbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZ29iYmxpbi9oaXZlL3dyaXRlci9IaXZlTWV0YWRhdGFXcml0ZXIuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...pache/gobblin/iceberg/writer/GobblinMCEWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvd3JpdGVyL0dvYmJsaW5NQ0VXcml0ZXIuamF2YQ==) | `74.01% <75.75%> (+1.65%)` | :arrow_up: |
| [...bblin/iceberg/writer/GobblinMetadataException.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvd3JpdGVyL0dvYmJsaW5NZXRhZGF0YUV4Y2VwdGlvbi5qYXZh) | `100.00% <100.00%> (ø)` | |
| [.../gobblin/iceberg/writer/IcebergMetadataWriter.java](https://codecov.io/gh/apache/gobblin/pull/3419/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) | `72.21% <100.00%> (+0.10%)` | :arrow_up: |
| [.../java/org/apache/gobblin/cluster/SleepingTask.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1jbHVzdGVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NsdXN0ZXIvU2xlZXBpbmdUYXNrLmphdmE=) | `39.39% <0.00%> (-6.07%)` | :arrow_down: |
| [...lin/restli/throttling/ZookeeperLeaderElection.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1yZXN0bGkvZ29iYmxpbi10aHJvdHRsaW5nLXNlcnZpY2UvZ29iYmxpbi10aHJvdHRsaW5nLXNlcnZpY2Utc2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3Jlc3RsaS90aHJvdHRsaW5nL1pvb2tlZXBlckxlYWRlckVsZWN0aW9uLmphdmE=) | `70.00% <0.00%> (-2.23%)` | :arrow_down: |
| [...main/java/org/apache/gobblin/yarn/YarnService.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi15YXJuL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3lhcm4vWWFyblNlcnZpY2UuamF2YQ==) | `15.46% <0.00%> (-0.78%)` | :arrow_down: |
| [...pache/gobblin/configuration/ConfigurationKeys.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1hcGkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2dvYmJsaW4vY29uZmlndXJhdGlvbi9Db25maWd1cmF0aW9uS2V5cy5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...apache/gobblin/salesforce/SalesforceExtractor.java](https://codecov.io/gh/apache/gobblin/pull/3419/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-Z29iYmxpbi1zYWxlc2ZvcmNlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3NhbGVzZm9yY2UvU2FsZXNmb3JjZUV4dHJhY3Rvci5qYXZh) | `0.00% <0.00%> (ø)` | |
| ... and [13 more](https://codecov.io/gh/apache/gobblin/pull/3419/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/3419?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/3419?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 [255bdc1...c6bcb1c](https://codecov.io/gh/apache/gobblin/pull/3419?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