fresh-borzoni commented on code in PR #2747:
URL: https://github.com/apache/fluss/pull/2747#discussion_r3321909458
##########
fluss-client/src/main/java/org/apache/fluss/client/write/RecordAccumulator.java:
##########
@@ -720,6 +729,28 @@ private List<ReadyWriteBatch>
drainBatchesForOneNode(Cluster cluster, Integer no
continue;
}
+ if (tableBucket.getTableId() != first.tableId()) {
Review Comment:
Only the first stale batch is aborted, then we continue to the next bucket -
n stale batches take n drain passes (and n WARN lines) to clear.
Shall we loop over the consecutive stale head batches and abort them in one
pass (outside the lock)?
##########
fluss-client/src/main/java/org/apache/fluss/client/write/RecordAccumulator.java:
##########
@@ -720,6 +729,28 @@ private List<ReadyWriteBatch>
drainBatchesForOneNode(Cluster cluster, Integer no
continue;
}
+ if (tableBucket.getTableId() != first.tableId()) {
+ LOG.warn(
+ "Table {} has been dropped and re-created with a
new table ID. Old ID: {}, New ID: {}. "
+ + "Aborting pending batches for the old
table instance.",
+ physicalTablePath,
+ first.tableId(),
+ tableBucket.getTableId());
+ TableNotExistException reason =
+ new TableNotExistException(
+ String.format(
+ "Table '%s' has been dropped and
re-created with a new table ID (old: %d, new: %d). "
+ + "Further writes to the
old table instance cannot proceed. "
+ + "Please recreate the
writer with the new table metadata.",
+ physicalTablePath,
+ first.tableId(),
+ tableBucket.getTableId()),
+ null,
+ true);
+ abortBatch(reason, deque.pollFirst());
Review Comment:
this fires the user callback while we're still holding the deque lock.
Shall we poll under the lock, then abort + deallocate after the synchronized
block?
--
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]
