wombatu-kun opened a new issue, #20315:
URL: https://github.com/apache/hudi/issues/20315
### Bug Description
**What happened:**
When a step fails after a Kafka Connect commit has already completed, the
sink writes the same Kafka records again in the next commit, so a COW table
ends up with duplicate rows. `KafkaConnectTransactionServices.endCommit`
completes the instant inside `javaClient.commit(...)` and then runs post-commit
actions: automatic cleaning, archiving and inline table services inside
`commitStats` (rethrown, since `hoodie.write.can.ignore.post.commit.failures`
defaults to false), then table service scheduling and `syncMeta()`. An
exception from any of them reaches
`ConnectTransactionCoordinator.onReceiveWriteStatus`, which treats the commit
as failed: it keeps its old Kafka offsets and skips the ACK. The next
START_COMMIT carries the offsets from before the completed commit, the
participants seek back to them, and the committed records are written into the
next instant. On COW, the default, the sink writes with bulk insert, so they
become duplicate rows, and the next commit can even record lo
wer Kafka offsets than the completed one.
**What you expected:**
A failure after the instant completed is logged and the commit counts as
successful, since its data and Kafka offsets are already visible; the failed
action runs again with the next commit.
**Steps to reproduce:**
1. Enable meta sync with a sync tool whose `syncHoodieTable()` throws,
standing in for an unreachable metastore: `hoodie.meta.sync.enable=true` and
`hoodie.meta.sync.classes=<failing tool>`.
2. Run the sink for a few commits.
3. Every completed instant contains the Kafka records from offset 0 again.
I reproduced this with the real `KafkaConnectTransactionServices` on a local
table, the real coordinator and participant, and the test helper
`MockKafkaConnect` replaying records from the offset the participant seeks to:
three completed instants wrote Kafka offsets `0..94`, `0..89` and `0..94`.
### Environment
**Hudi version:** master (1.3.0-SNAPSHOT)
**Query engine:** Kafka Connect (Hudi sink)
**Relevant configs:** `hoodie.meta.sync.enable=true` for the meta sync
trigger; the cleaning and archiving triggers need no config, both run after
every commit by default
### Logs and Stack Trace
From the reproduction above:
```
ERROR org.apache.hudi.connect.transaction.ConnectTransactionCoordinator [] -
Fatal error while committing file
org.apache.hudi.exception.HoodieMetaSyncException: Could not sync using the
meta sync class
org.apache.hudi.writers.TestKafkaConnectTransactionServices$FailingMetaSyncTool
at
org.apache.hudi.sync.common.util.SyncUtilHelpers.runHoodieMetaSync(SyncUtilHelpers.java:106)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
