nzw921rx commented on PR #10849:
URL: https://github.com/apache/seatunnel/pull/10849#issuecomment-4394024678
Race condition in BigtableSinkWriter: concurrent write() and prepareCommit()
may cause duplicate writes
When a record causes buffer.size() >= batchMutationSize exactly as a
checkpoint fires prepareCommit(), both threads pass the !buffer.isEmpty() guard
and call bulkMutate() on the same list concurrently — resulting in duplicate
writes. The subsequent concurrent buffer.clear() calls also risk
ConcurrentModificationException.
The fix should follow the SeaTunnel convention used by other connectors
(e.g. InfluxDB sink) — synchronize only on the buffer object to keep
convertRowToMutation() outside the critical section, and clear before mutate to
avoid re-sending on exception:
```java
@Override
public void write(SeaTunnelRow element) throws IOException {
RowKeyMutation entry = convertRowToMutation(element); // outside lock:
no shared state
synchronized (buffer) {
buffer.add(entry);
if (buffer.size() >= parameters.getBatchMutationSize()) {
flush();
}
}
}
private void flush() {
synchronized (buffer) {
if (buffer.isEmpty()) {
return;
}
List<RowKeyMutation> toFlush = new ArrayList<>(buffer);
buffer.clear(); // clear first: prevents re-sending if bulkMutate
throws
bigtableClient.bulkMutate(toFlush);
}
}
```
--
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]
