platinumhamburg commented on code in PR #3651:
URL: https://github.com/apache/fluss/pull/3651#discussion_r3587018226
##########
fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java:
##########
@@ -473,70 +475,63 @@ public void shutdown() {
logsByDataDir.computeIfAbsent(dataDir, ignored -> new
ArrayList<>()).add(logTablet);
}
- List<LogShutdownTask> shutdownTasks = new ArrayList<>();
+ Map<File, CompletableFuture<Void>> closingFuturesByDataDir = new
LinkedHashMap<>();
for (Map.Entry<File, List<LogTablet>> entry :
logsByDataDir.entrySet()) {
- shutdownTasks.add(shutdownLogsInDir(entry.getKey(),
entry.getValue()));
+ File dataDir = entry.getKey();
+ List<LogTablet> logs = entry.getValue();
+ String dataDirAbsolutePath = dataDir.getAbsolutePath();
+ LOG.info("Shutting down {} logs in dir {}", logs.size(),
dataDirAbsolutePath);
+ closingFuturesByDataDir.put(
+ dataDir,
+ closeTabletsConcurrently(
+ logs,
+ "log-tablet-closing-" + dataDirAbsolutePath,
+ this::closeLogTablet));
}
- try {
- for (LogShutdownTask shutdownTask : shutdownTasks) {
- waitForShutdownLogsInDir(shutdownTask);
- }
- } finally {
- for (LogShutdownTask shutdownTask : shutdownTasks) {
- shutdownTask.pool.shutdown();
- }
+ for (Map.Entry<File, List<LogTablet>> entry :
logsByDataDir.entrySet()) {
+ waitForShutdownLogsInDir(
+ entry.getKey(), entry.getValue(),
closingFuturesByDataDir.get(entry.getKey()));
}
LOG.info("Shut down LogManager complete.");
}
- private LogShutdownTask shutdownLogsInDir(File dataDir, List<LogTablet>
logs) {
- String dataDirAbsolutePath = dataDir.getAbsolutePath();
- LOG.info("Shutting down {} logs in dir {}", logs.size(),
dataDirAbsolutePath);
-
- List<Future<?>> jobsForTabletDir = new ArrayList<>();
- ExecutorService pool = createThreadPool("log-tablet-closing-" +
dataDirAbsolutePath);
- for (LogTablet logTablet : logs) {
- Runnable runnable =
- () -> {
- try {
- logTablet.flush(true);
- logTablet.close();
- } catch (IOException e) {
- throw new FlussRuntimeException(e);
- }
- };
- jobsForTabletDir.add(pool.submit(runnable));
+ private void closeLogTablet(LogTablet logTablet) {
+ try {
+ logTablet.flush(true);
+ logTablet.close();
+ } catch (IOException e) {
+ throw new FlussRuntimeException(e);
}
- return new LogShutdownTask(dataDir, logs, pool, jobsForTabletDir);
}
- private void waitForShutdownLogsInDir(LogShutdownTask shutdownTask) {
- for (Future<?> future : shutdownTask.jobs) {
- try {
- future.get();
- } catch (InterruptedException e) {
- LOG.warn("Interrupted while shutting down LogManager.");
- } catch (ExecutionException e) {
- LOG.warn("There was an error in one of the threads during
LogManager shutdown", e);
- }
+ private void waitForShutdownLogsInDir(
Review Comment:
Could we preserve each per-tablet close failure here?
`CompletableFuture.allOf()` completes exceptionally if any child fails, but
`join()` exposes only one exceptional cause. Before this change,
`waitForShutdownLogsInDir` iterated over every `Future` and logged each
`ExecutionException`.
If multiple `LogTablet`s in the same data directory fail to close, the new
code logs only one failure and loses the other affected `TableBucket`s, which
is an observability regression during shutdown. The close tasks still all run
and the clean-shutdown marker decision remains correct, so this is non-blocking.
Logging failures with the `TableBucket` inside each tablet's close action,
or retaining the child futures and aggregating all causes after `allOf`, would
preserve the previous diagnostics.
##########
fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java:
##########
@@ -206,13 +206,7 @@ public void shutdown() {
LOG.info("Shutting down KvManager");
isShutdown = true;
List<KvTablet> kvs = new ArrayList<>(currentKvs.values());
- for (KvTablet kvTablet : kvs) {
- try {
- kvTablet.close();
- } catch (Exception e) {
- LOG.warn("Exception while closing kv tablet {}.",
kvTablet.getTableBucket(), e);
- }
- }
+ closeTabletsConcurrently(kvs, "kv-tablet-closing",
this::closeKvTablet).join();
Review Comment:
Could we confirm that all users of the KV tablets have been quiesced before
starting this parallel close?
`TabletServer.stopServices()` starts `rpcServer.closeAsync()` but does not
await it before calling `kvManager.shutdown()`. `TabletService.shutdown()` is
currently a no-op, and `ReplicaManager.shutdown()`—which stops snapshot,
fetcher, and delayed-operation activity—runs only after the KV and Log managers
have already shut down.
This race already existed with serial close, but this change can move up to
`recoveryThreads` tablets into the closing state at once, potentially
increasing the number of queued or in-flight operations that observe a closed
RocksDB.
If pending operations are allowed to fail during server shutdown, could we
document or test that contract? If graceful draining is intended, the RPC and
Replica users should be awaited or otherwise quiesced before this call. An
integration test with an in-flight put, fetch, or snapshot during TabletServer
shutdown would make the intended behavior explicit.
--
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]
