swuferhong commented on code in PR #3651:
URL: https://github.com/apache/fluss/pull/3651#discussion_r3587780859
##########
fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java:
##########
@@ -670,37 +670,47 @@ private WalBuilder createWalBuilder(int schemaId, RowType
rowType) throws Except
public void flush(long exclusiveUpToLogOffset, FatalErrorHandler
fatalErrorHandler) {
// todo: need to introduce a backpressure mechanism
// to avoid too much records in kvPreWriteBuffer
- inWriteLock(
- kvLock,
- () -> {
- // when kv manager is closed which means kv tablet is
already closed,
- // but the tablet server may still handle fetch log
request from follower
- // as the tablet rpc service is closed asynchronously,
then update the watermark
- // and then flush the pre-write buffer.
-
- // In such case, if the tablet is already closed, we won't
flush pre-write
- // buffer, just warning it.
- if (isClosed) {
- LOG.warn(
- "The kv tablet for {} is already closed,
ignore flushing kv pre-write buffer.",
- tableBucket);
- } else {
- try {
- int rowCountDiff =
kvPreWriteBuffer.flush(exclusiveUpToLogOffset);
- if (exclusiveUpToLogOffset > flushedLogOffset) {
- flushedLogOffset = exclusiveUpToLogOffset;
+ Throwable flushError =
+ inWriteLock(
+ kvLock,
+ () -> {
+ // when kv manager is closed which means kv tablet
is already closed,
+ // but the tablet server may still handle fetch
log request from
+ // follower
+ // as the tablet rpc service is closed
asynchronously, then update the
+ // watermark and then flush the pre-write buffer.
+
+ // In such case, if the tablet is already closed,
we won't flush
+ // pre-write buffer, just warning it.
+ if (isClosed) {
+ LOG.warn(
+ "The kv tablet for {} is already
closed, ignore flushing kv pre-write buffer.",
+ tableBucket);
+ return null;
}
- if (rowCount != ROW_COUNT_DISABLED) {
- // row count is enabled, we update the row
count after flush.
- long currentRowCount = rowCount;
- rowCount = currentRowCount + rowCountDiff;
+
+ try {
+ int rowCountDiff =
kvPreWriteBuffer.flush(exclusiveUpToLogOffset);
+ if (exclusiveUpToLogOffset > flushedLogOffset)
{
+ flushedLogOffset = exclusiveUpToLogOffset;
+ }
+ if (rowCount != ROW_COUNT_DISABLED) {
+ // row count is enabled, we update the row
count after flush.
+ long currentRowCount = rowCount;
+ rowCount = currentRowCount + rowCountDiff;
+ }
+ return null;
+ } catch (Throwable t) {
+ return t;
}
- } catch (Throwable t) {
- fatalErrorHandler.onFatalError(
- new KvStorageException("Failed to flush kv
pre-write buffer."));
- }
- }
- });
+ });
+
+ if (flushError != null) {
+ // The fatal error handler may synchronously shut down the server,
which closes this
+ // tablet from another thread. Invoke it after releasing kvLock to
avoid deadlock.
+ fatalErrorHandler.onFatalError(
Review Comment:
Good idea!
--
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]
