swuferhong commented on code in PR #3651:
URL: https://github.com/apache/fluss/pull/3651#discussion_r3584123251
##########
fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java:
##########
@@ -473,69 +475,55 @@ 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 (Exception e) {
+ LOG.warn("Exception while closing log tablet {}.",
logTablet.getTableBucket(), e);
Review Comment:
Thanks @fresh-borzoni, good catch. The old implementation also wrote the
clean-shutdown marker after catching an ExecutionException, so the issue
predates this PR. However, this refactor currently swallows the exception even
earlier. I’ll preserve the failure on the per-directory future and only write
the marker when all close tasks for that directory complete successfully.
--
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]
