fresh-borzoni commented on code in PR #3651:
URL: https://github.com/apache/fluss/pull/3651#discussion_r3584004363
##########
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:
If flush fails here, we still write the clean-shutdown marker below and
next startup skips log recovery because of it.
Shouldn't a failed flush block the marker for that dir?
The helper already fails the future when the action throws, so it's probably
this: don't catch here, catch at the join and skip the marker, wdyt?
--
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]
