MartijnVisser commented on code in PR #29445:
URL: https://github.com/apache/flink/pull/29445#discussion_r4245873361
##########
flink-state-backends/flink-statebackend-common/src/main/java/org/apache/flink/state/common/PeriodicMaterializationManager.java:
##########
@@ -205,12 +205,26 @@ public void start() {
}
// task thread and task canceler can access this method
- public synchronized void close() {
+ public void close() {
LOG.info("Shutting down PeriodicMaterializationManager.");
- if (!periodicExecutor.isShutdown()) {
- periodicExecutor.shutdownNow();
+ synchronized (this) {
+ if (!periodicExecutor.isShutdown()) {
+ periodicExecutor.shutdownNow();
+ }
+ }
+
+ // A running callback may need the manager lock to schedule its next
materialization.
+ try {
+ if (!periodicExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
Review Comment:
`PeriodicMaterializationManagerTest` doesn't exercise this wait:
`ManuallyTriggeredScheduledExecutorService.awaitTermination` always returns
true, so nothing fails without the change.
##########
flink-state-backends/flink-statebackend-common/src/main/java/org/apache/flink/state/common/PeriodicMaterializationManager.java:
##########
@@ -205,12 +205,26 @@ public void start() {
}
// task thread and task canceler can access this method
- public synchronized void close() {
+ public void close() {
Review Comment:
After FLINK-40921 the changelog module still leaves 138 managers unclosed,
126 from `StateBackendMigrationTestBase`, which disposes without closing. With
operator id `test_op` they fire after 254.6 s, the same timer the CI crashes
hit.
##########
flink-state-backends/flink-statebackend-common/src/main/java/org/apache/flink/state/common/PeriodicMaterializationManager.java:
##########
@@ -205,12 +205,26 @@ public void start() {
}
// task thread and task canceler can access this method
- public synchronized void close() {
+ public void close() {
LOG.info("Shutting down PeriodicMaterializationManager.");
- if (!periodicExecutor.isShutdown()) {
- periodicExecutor.shutdownNow();
+ synchronized (this) {
+ if (!periodicExecutor.isShutdown()) {
+ periodicExecutor.shutdownNow();
+ }
+ }
+
+ // A running callback may need the manager lock to schedule its next
materialization.
Review Comment:
In production the scheduler only enqueues a mail, so this never waits for a
snapshot. Only the `SyncMailboxExecutor` in tests runs the materialization on
this thread.
--
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]
