From 7358775fb81ce9a9426a7e9ccbe3f8d3db445831 Mon Sep 17 00:00:00 2001 From: Yang Guo Date: Thu, 23 Jul 2026 20:16:27 +0800 Subject: [PATCH 1/2] fix zk notification purge race --- .../ZkNodeChangeNotificationWatcher.java | 10 ++- .../ZkNodeChangeNotificationWatcherTest.java | 71 +++++++++++++++++++ 2 files changed, 79 insertions(+), 2 deletions(-) diff --git a/fluss-server/src/main/java/org/apache/fluss/server/authorizer/ZkNodeChangeNotificationWatcher.java b/fluss-server/src/main/java/org/apache/fluss/server/authorizer/ZkNodeChangeNotificationWatcher.java index 31e6666ac9..f8fad3d610 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/authorizer/ZkNodeChangeNotificationWatcher.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/authorizer/ZkNodeChangeNotificationWatcher.java @@ -139,8 +139,14 @@ private void purgeObsoleteNotifications(long now, List sortedNotificatio for (String notification : sortedNotifications) { String notificationNode = seqNodeRoot + "/" + notification; try { - Stat state = zooKeeperClient.getStat(notificationNode).get(); - if (now - state.getCtime() >= changeExpirationMs) { + Optional state = zooKeeperClient.getStat(notificationNode); + if (!state.isPresent()) { + LOG.debug( + "Notification {} has already been purged by another watcher", + notificationNode); + continue; + } + if (now - state.get().getCtime() >= changeExpirationMs) { LOG.debug("Purging change notification {}", notificationNode); zooKeeperClient.deletePath(notificationNode); } diff --git a/fluss-server/src/test/java/org/apache/fluss/server/authorizer/ZkNodeChangeNotificationWatcherTest.java b/fluss-server/src/test/java/org/apache/fluss/server/authorizer/ZkNodeChangeNotificationWatcherTest.java index d43784c1ae..4b6a1aadc9 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/authorizer/ZkNodeChangeNotificationWatcherTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/authorizer/ZkNodeChangeNotificationWatcherTest.java @@ -25,6 +25,11 @@ import org.apache.fluss.testutils.common.AllCallbackWrapper; import org.apache.fluss.utils.clock.ManualClock; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.core.LogEvent; +import org.apache.logging.log4j.core.Logger; +import org.apache.logging.log4j.core.appender.AbstractAppender; +import org.apache.logging.log4j.core.config.Property; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; @@ -32,12 +37,16 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.List; +import java.util.Optional; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import static org.apache.fluss.testutils.common.CommonTestUtils.retry; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.spy; /** Test for {@link ZkNodeChangeNotificationWatcher }. */ public class ZkNodeChangeNotificationWatcherTest { @@ -119,6 +128,55 @@ void testZkNodeChangeNotifications() throws Exception { () -> assertThat(zookeeperClient.getChildren(seqNodeRoot)).hasSize(2)); } + @Test + void testNotificationDeletedByAnotherWatcherDuringPurge() throws Exception { + String seqNodeRoot = ZkData.AclChangesNode.path(); + String seqNodePrefix = ZkData.AclChangeNotificationNode.prefix(); + ZooKeeperClient zooKeeperClient = + ZOO_KEEPER_EXTENSION_WRAPPER + .getCustomExtension() + .getZooKeeperClient(NOPErrorHandler.INSTANCE); + Resource resource = Resource.cluster(); + zooKeeperClient.insertAclChangeNotification(resource); + + ZooKeeperClient spyingZooKeeperClient = spy(zooKeeperClient); + doAnswer( + invocation -> { + zooKeeperClient.deletePath(invocation.getArgument(0)); + return Optional.empty(); + }) + .when(spyingZooKeeperClient) + .getStat(anyString()); + + TestingNotificationHandler handler = new TestingNotificationHandler(); + ZkNodeChangeNotificationWatcher watcher = + new ZkNodeChangeNotificationWatcher( + spyingZooKeeperClient, + seqNodeRoot, + seqNodePrefix, + Duration.ofMinutes(5).toMillis(), + handler, + new ManualClock()); + + Logger logger = (Logger) LogManager.getLogger(ZkNodeChangeNotificationWatcher.class); + TestingAppender appender = new TestingAppender(); + appender.start(); + logger.addAppender(appender); + try { + watcher.start(); + assertThat(handler.resourceNotifications).containsExactly(resource); + assertThat(appender.messages) + .noneMatch( + message -> + message.contains( + "Error while purging obsolete notification change")); + } finally { + watcher.stop(); + logger.removeAppender(appender); + appender.stop(); + } + } + private static class TestingNotificationHandler implements ZkNodeChangeNotificationWatcher.NotificationHandler { public BlockingQueue resourceNotifications = new LinkedBlockingQueue<>(); @@ -129,4 +187,17 @@ public void processNotification(byte[] notification) { resourceNotifications.add(resource); } } + + private static class TestingAppender extends AbstractAppender { + private final List messages = new ArrayList<>(); + + private TestingAppender() { + super("testing", null, null, true, Property.EMPTY_ARRAY); + } + + @Override + public void append(LogEvent event) { + messages.add(event.getMessage().getFormattedMessage()); + } + } } From d67bd338db7f83db961f4829f9f549ffc0a00561 Mon Sep 17 00:00:00 2001 From: Yang Guo Date: Fri, 24 Jul 2026 11:10:46 +0800 Subject: [PATCH 2/2] [server] Strengthen notification purge race test --- .../ZkNodeChangeNotificationWatcherTest.java | 59 ++++++++++++------- 1 file changed, 37 insertions(+), 22 deletions(-) diff --git a/fluss-server/src/test/java/org/apache/fluss/server/authorizer/ZkNodeChangeNotificationWatcherTest.java b/fluss-server/src/test/java/org/apache/fluss/server/authorizer/ZkNodeChangeNotificationWatcherTest.java index 4b6a1aadc9..f5707c3db2 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/authorizer/ZkNodeChangeNotificationWatcherTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/authorizer/ZkNodeChangeNotificationWatcherTest.java @@ -25,10 +25,13 @@ import org.apache.fluss.testutils.common.AllCallbackWrapper; import org.apache.fluss.utils.clock.ManualClock; +import org.apache.logging.log4j.Level; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.core.LogEvent; -import org.apache.logging.log4j.core.Logger; +import org.apache.logging.log4j.core.LoggerContext; import org.apache.logging.log4j.core.appender.AbstractAppender; +import org.apache.logging.log4j.core.config.Configuration; +import org.apache.logging.log4j.core.config.LoggerConfig; import org.apache.logging.log4j.core.config.Property; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; @@ -37,16 +40,12 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.List; -import java.util.Optional; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import static org.apache.fluss.testutils.common.CommonTestUtils.retry; import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.ArgumentMatchers.anyString; -import static org.mockito.Mockito.doAnswer; -import static org.mockito.Mockito.spy; /** Test for {@link ZkNodeChangeNotificationWatcher }. */ public class ZkNodeChangeNotificationWatcherTest { @@ -139,40 +138,56 @@ void testNotificationDeletedByAnotherWatcherDuringPurge() throws Exception { Resource resource = Resource.cluster(); zooKeeperClient.insertAclChangeNotification(resource); - ZooKeeperClient spyingZooKeeperClient = spy(zooKeeperClient); - doAnswer( - invocation -> { - zooKeeperClient.deletePath(invocation.getArgument(0)); - return Optional.empty(); - }) - .when(spyingZooKeeperClient) - .getStat(anyString()); - - TestingNotificationHandler handler = new TestingNotificationHandler(); + TestingNotificationHandler handler = + new TestingNotificationHandler() { + @Override + public void processNotification(byte[] notification) { + super.processNotification(notification); + try { + for (String child : zooKeeperClient.getChildren(seqNodeRoot)) { + zooKeeperClient.deletePath(seqNodeRoot + "/" + child); + } + } catch (Exception e) { + throw new RuntimeException(e); + } + } + }; ZkNodeChangeNotificationWatcher watcher = new ZkNodeChangeNotificationWatcher( - spyingZooKeeperClient, + zooKeeperClient, seqNodeRoot, seqNodePrefix, Duration.ofMinutes(5).toMillis(), handler, new ManualClock()); - Logger logger = (Logger) LogManager.getLogger(ZkNodeChangeNotificationWatcher.class); TestingAppender appender = new TestingAppender(); appender.start(); - logger.addAppender(appender); + LoggerContext loggerContext = (LoggerContext) LogManager.getContext(false); + Configuration configuration = loggerContext.getConfiguration(); + LoggerConfig loggerConfig = + new LoggerConfig( + ZkNodeChangeNotificationWatcher.class.getName(), Level.DEBUG, false); + loggerConfig.addAppender(appender, Level.DEBUG, null); + configuration.addLogger(loggerConfig.getName(), loggerConfig); + loggerContext.updateLoggers(); try { watcher.start(); assertThat(handler.resourceNotifications).containsExactly(resource); assertThat(appender.messages) - .noneMatch( + .anyMatch( message -> - message.contains( - "Error while purging obsolete notification change")); + message.startsWith( + "Notification " + + seqNodeRoot + + "/" + + seqNodePrefix) + && message.endsWith( + " has already been purged by another watcher")); } finally { watcher.stop(); - logger.removeAppender(appender); + configuration.removeLogger(loggerConfig.getName()); + loggerContext.updateLoggers(); appender.stop(); } }