From d2fcec2dfd00de2e4fe40b0ec71f60acfffb630a Mon Sep 17 00:00:00 2001 From: Dmitry Kasimovskiy Date: Wed, 23 Sep 2026 15:08:05 +0300 Subject: [PATCH 1/7] fix(tarantool-core): send IPROTO_WATCH after authentication --- .gitlab-ci.yml | 14 --------- CHANGELOG.md | 10 +++++++ .../io/tarantool/core/IProtoClientImpl.java | 29 +++++++++++-------- 3 files changed, 27 insertions(+), 26 deletions(-) diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml index 75eb8be9..c92baabf 100644 --- a/.gitlab-ci.yml +++ b/.gitlab-ci.yml @@ -1,5 +1,4 @@ stages: - - ee_tests - package variables: @@ -35,19 +34,6 @@ cache: KUBERNETES_SERVICE_MEMORY_REQUEST: "2Gi" KUBERNETES_SERVICE_MEMORY_LIMIT: "10Gi" -# Trigger `Tarantool Java SDK` enterprise tests run -run_ee_tests: - stage: ee_tests - allow_failure: true - trigger: - project: tarantool/java/tarantool-java-sdk-ee-testing - strategy: depend - branch: master - variables: - MAIN_REPO_SHA: "${CI_COMMIT_SHA}" - MAIN_REPO_COMMIT_REF_NAME: "${CI_COMMIT_REF_NAME}" - PARENT_PIPELINE_ID: "${CI_PIPELINE_ID}" - run_ee_release: stage: package allow_failure: true diff --git a/CHANGELOG.md b/CHANGELOG.md index 658eb0f6..99201cbf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,16 @@ - Support ISO 8601 duration parsing and formatting for `Interval` +### Bug fixes + +- Register watchers (including the automatic `box.shutdown` watcher created when + `gracefulShutdown` is enabled) only after the session is authenticated, i.e. after + `authorize()` or `ping()` completes, instead of right after the greeting. Tarantool EE + builds with option `security.disable_guest: true` allow only `auth`, `ping`, `id` and + `vote` requests before authentication and answered the early `IPROTO_WATCH` with + `ER_AUTH_REQUIRED`; since 1.7.0 that watcher error failed the whole connect procedure + and put the connection into an endless reconnect loop. + ## [1.7.1] - 2026-08-31 ### Dependencies diff --git a/tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java b/tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java index c7d9b27a..f1aebc26 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java +++ b/tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java @@ -185,20 +185,13 @@ public CompletableFuture connect(InetSocketAddress address, long timeoutMs public CompletableFuture connect( InetSocketAddress address, long timeoutMs, boolean gracefulShutdown) { if (gracefulShutdown) { - // it does not send watch message if connection is not connected, - // it sends immediately after successful connect + // the watch request is sent later, after authorize() or ping() watch(SHUTDOWN_EVENT_KEY, this::shutdownEventCallback); } serverProtocolVersion = new CompletableFuture<>(); serverFeatures = new CompletableFuture<>(); - return connection - .connect(address, timeoutMs) - .thenRun( - () -> { - updateWatchers(); - updateServerInfo(); - }); + return connection.connect(address, timeoutMs).thenRun(this::updateServerInfo); } @Override @@ -227,7 +220,13 @@ public CompletableFuture authorize( promise.completeExceptionally(new ClientException("No greeting, connect firstly!")); return promise; } - return runRequest(new IProtoAuth(user, password, greeting.get().getSalt(), authType), opts); + // Tarantool EE rejects IPROTO_WATCH sent before IPROTO_AUTH with ER_AUTH_REQUIRED + return runRequest(new IProtoAuth(user, password, greeting.get().getSalt(), authType), opts) + .thenApply( + response -> { + updateWatchers(); + return response; + }); } @Override @@ -617,12 +616,18 @@ public CompletableFuture execute( @Override public CompletableFuture ping() { - return runRequest(new IProtoPing(), DEFAULT_REQUEST_OPTS); + return ping(DEFAULT_REQUEST_OPTS); } @Override public CompletableFuture ping(IProtoRequestOpts opts) { - return runRequest(new IProtoPing(), opts); + // guest connect path: ping is allowed before authentication + return runRequest(new IProtoPing(), opts) + .thenApply( + response -> { + updateWatchers(); + return response; + }); } @Override From d8ab5140ef0817aeaa99aeae52ca7975725fcb12 Mon Sep 17 00:00:00 2001 From: Dmitry Kasimovskiy Date: Wed, 23 Sep 2026 17:09:45 +0300 Subject: [PATCH 2/7] fix(test): call ping() after connect() in WatchersTest IProtoClientWatchersTest uses connect() without auth. Since watchers are now sent after ping() (TNTP-10388 fix), ping must be called to trigger watcher registration. --- .../io/tarantool/core/integration/IProtoClientWatchersTest.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientWatchersTest.java b/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientWatchersTest.java index e4c9aa3e..66ba5199 100644 --- a/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientWatchersTest.java +++ b/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientWatchersTest.java @@ -58,6 +58,8 @@ private IProtoClient getClientAndConnect(TarantoolContainer tt) throws Except InetSocketAddress address = tt.mappedAddress(); IProtoClient client = new IProtoClientImpl(factory, factory.getTimerService()); client.connect(address, 3_000).get(); + // Complete the handshake with ping so that watchers are sent + client.ping().get(); return client; } From d3541720d828601c3bbcf943752f873cc8150600 Mon Sep 17 00:00:00 2001 From: Dmitry Kasimovskiy Date: Thu, 24 Sep 2026 10:00:45 +0300 Subject: [PATCH 3/7] fix(test): adapt integration tests to post-auth watcher registration - testAuthorization: auth response syncId is now 2 since IPROTO_WATCH is no longer allocated before the auth request (TNTP-10388) - testWatcherRecoveryAfterReconnect: call ping() after reconnect to re-trigger watcher registration, adjust sleep timings --- .../io/tarantool/core/integration/IProtoClientTest.java | 3 ++- .../core/integration/IProtoClientWatchersTest.java | 9 +++++++-- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientTest.java b/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientTest.java index f6ca2476..46e18737 100644 --- a/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientTest.java +++ b/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientTest.java @@ -83,6 +83,7 @@ public class IProtoClientTest extends BaseTest { private static final IProtoRequestOpts DEFAULT_REQUEST_OPTS = IProtoRequestOpts.empty().withRequestTimeout(5000); + private static final long AUTH_SYNC_ID = 2; private static TarantoolContainer tt; private static int spaceAId; private static int spaceBId; @@ -1386,7 +1387,7 @@ public void testOperationsWithPushHandler() throws Exception { public void testAuthorization() throws Exception { IProtoClient client = createClientAndConnect(address, true); IProtoMessage message = client.authorize("user_a", "secret_a").get(); - checkMessageHeader(message, IPROTO_OK, 3); + checkMessageHeader(message, IPROTO_OK, AUTH_SYNC_ID); assertEquals(0, message.getBody().asMapValue().map().size()); } diff --git a/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientWatchersTest.java b/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientWatchersTest.java index 66ba5199..8eaa4721 100644 --- a/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientWatchersTest.java +++ b/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientWatchersTest.java @@ -179,12 +179,17 @@ private void testWatcherRecoveryAfterReconnectOnContainer( "box.broadcast('keyA', 'myEvent');" + "box.broadcast('keyB', {1, 2, 3});" + "box.broadcast('keyC', 'wontbecaught');"); + // let the first broadcast events arrive before the connection is closed Thread.sleep(100); client.close(); - Thread.sleep(100); + Thread.sleep(200); InetSocketAddress address = tt.mappedAddress(); client.connect(address, 3_000).get(); - Thread.sleep(1000); + // Re-trigger watcher registration after reconnect (same as getClientAndConnect) + client.ping().get(); + // Give time for handleClose to finish clearing stateContext on the Netty thread + // and for IPROTO_WATCH responses to arrive before sending new broadcast + Thread.sleep(1500); TarantoolContainerClientHelper.executeCommand( tt, From c379598ada1ff5b53f8f88d5f929f53b25ec7ab6 Mon Sep 17 00:00:00 2001 From: Dmitry Kasimovskiy Date: Thu, 24 Sep 2026 15:52:58 +0300 Subject: [PATCH 4/7] fix(tarantool-core): rewrite set watchers logic Fallback to set watchers after authentication if ER_AUTH_REQUIRED happens on connect. --- .../io/tarantool/core/IProtoClientImpl.java | 45 ++++++++++++++----- .../protocol/fsm/WatcherStateMachine.java | 33 +++++++++++++- .../core/integration/IProtoClientTest.java | 3 +- .../integration/IProtoClientWatchersTest.java | 11 +---- 4 files changed, 68 insertions(+), 24 deletions(-) diff --git a/tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java b/tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java index f1aebc26..b248c732 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java +++ b/tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java @@ -101,6 +101,7 @@ public class IProtoClientImpl implements IProtoClient { private final WatcherOptions watcherOpts; private CompletableFuture serverProtocolVersion; private CompletableFuture> serverFeatures; + private volatile boolean authorized; private Set clientFeaturesEnum; private List clientFeaturesList; private LongTaskTimer requestTimer; @@ -185,13 +186,19 @@ public CompletableFuture connect(InetSocketAddress address, long timeoutMs public CompletableFuture connect( InetSocketAddress address, long timeoutMs, boolean gracefulShutdown) { if (gracefulShutdown) { - // the watch request is sent later, after authorize() or ping() watch(SHUTDOWN_EVENT_KEY, this::shutdownEventCallback); } + authorized = false; serverProtocolVersion = new CompletableFuture<>(); serverFeatures = new CompletableFuture<>(); - return connection.connect(address, timeoutMs).thenRun(this::updateServerInfo); + return connection + .connect(address, timeoutMs) + .thenRun( + () -> { + updateWatchers(); + updateServerInfo(); + }); } @Override @@ -220,10 +227,11 @@ public CompletableFuture authorize( promise.completeExceptionally(new ClientException("No greeting, connect firstly!")); return promise; } - // Tarantool EE rejects IPROTO_WATCH sent before IPROTO_AUTH with ER_AUTH_REQUIRED + // re-register watchers rejected before authentication return runRequest(new IProtoAuth(user, password, greeting.get().getSalt(), authType), opts) .thenApply( response -> { + authorized = true; updateWatchers(); return response; }); @@ -621,13 +629,7 @@ public CompletableFuture ping() { @Override public CompletableFuture ping(IProtoRequestOpts opts) { - // guest connect path: ping is allowed before authentication - return runRequest(new IProtoPing(), opts) - .thenApply( - response -> { - updateWatchers(); - return response; - }); + return runRequest(new IProtoPing(), opts); } @Override @@ -812,7 +814,13 @@ private synchronized void updateWatchers() { long syncId = allocateSyncIds(1); fsm = new WatcherStateMachine( - watcherEntry.getKey(), syncId, watcher, connection, watcherOpts, timerService); + watcherEntry.getKey(), + syncId, + watcher, + connection, + watcherOpts, + timerService, + failed -> onWatchAuthRequired(watcher, failed)); watcher.setStateContext(fsm); watcher.setSyncId(syncId); fsmRegistry.put(syncId, fsm); @@ -836,6 +844,21 @@ private synchronized void updateWatchers() { } } + /** + * Resets a watcher rejected before authentication, so that it is re-registered after authorize() + * or immediately, if the client is already authorized. + */ + private synchronized void onWatchAuthRequired(Watcher watcher, WatcherStateMachine failed) { + fsmRegistry.remove(failed.getSyncId()); + // a newer registration attempt may already be active + if (watcher.getStateContext() == failed) { + watcher.setStateContext(null); + } + if (authorized) { + updateWatchers(); + } + } + protected long allocateSyncIds(int count) { // n + count < 0 is a check for overflow return syncIdSequence.updateAndGet(n -> (n + count) < 0 ? count : (n + count)); diff --git a/tarantool-core/src/main/java/io/tarantool/core/protocol/fsm/WatcherStateMachine.java b/tarantool-core/src/main/java/io/tarantool/core/protocol/fsm/WatcherStateMachine.java index f4e6a986..8c889092 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/protocol/fsm/WatcherStateMachine.java +++ b/tarantool-core/src/main/java/io/tarantool/core/protocol/fsm/WatcherStateMachine.java @@ -23,6 +23,9 @@ public class WatcherStateMachine implements IProtoStateMachine { + /** Tarantool error code for requests rejected before authentication (EE only). */ + public static final int ER_AUTH_REQUIRED = 258; + private final Connection connection; private static final Logger log = LoggerFactory.getLogger(WatcherStateMachine.class); @@ -37,8 +40,12 @@ public class WatcherStateMachine implements IProtoStateMachine { private final WatcherOptions opts; + private final Consumer onAuthRequired; + private final CompletableFuture registered = new CompletableFuture<>(); + private final long syncId; + private boolean calledOnce; public WatcherStateMachine( @@ -48,13 +55,26 @@ public WatcherStateMachine( Connection connection, WatcherOptions opts, Timer timerService) { + this(key, syncId, callback, connection, opts, timerService, null); + } + + public WatcherStateMachine( + String key, + long syncId, + Consumer callback, + Connection connection, + WatcherOptions opts, + Timer timerService, + Consumer onAuthRequired) { this.key = key; + this.syncId = syncId; this.connection = connection; this.callback = callback; this.request = new IProtoWatch(key); this.request.setSyncId(syncId); this.opts = opts; this.timerService = timerService; + this.onAuthRequired = onAuthRequired; } @Override @@ -92,8 +112,13 @@ public boolean process(IProtoResponse message) { } else { ClientException error = new ClientException("watcher error: %s", message); registered.completeExceptionally(error); - log.warn("got error for watcher: {}", message); - opts.getErrorHandler().accept(key, error); + if (onAuthRequired != null && message.getErrorCode() == ER_AUTH_REQUIRED) { + log.debug("watcher '{}' rejected before authentication, deferred", key); + onAuthRequired.accept(this); + } else { + log.warn("got error for watcher: {}", message); + opts.getErrorHandler().accept(key, error); + } } return false; @@ -110,6 +135,10 @@ public CompletableFuture registered() { return registered; } + public long getSyncId() { + return syncId; + } + private void onSendComplete(Void r, Throwable exc) { if (exc != null) { log.warn("could not send IPROTO_WATCH packet", exc); diff --git a/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientTest.java b/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientTest.java index 46e18737..f6ca2476 100644 --- a/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientTest.java +++ b/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientTest.java @@ -83,7 +83,6 @@ public class IProtoClientTest extends BaseTest { private static final IProtoRequestOpts DEFAULT_REQUEST_OPTS = IProtoRequestOpts.empty().withRequestTimeout(5000); - private static final long AUTH_SYNC_ID = 2; private static TarantoolContainer tt; private static int spaceAId; private static int spaceBId; @@ -1387,7 +1386,7 @@ public void testOperationsWithPushHandler() throws Exception { public void testAuthorization() throws Exception { IProtoClient client = createClientAndConnect(address, true); IProtoMessage message = client.authorize("user_a", "secret_a").get(); - checkMessageHeader(message, IPROTO_OK, AUTH_SYNC_ID); + checkMessageHeader(message, IPROTO_OK, 3); assertEquals(0, message.getBody().asMapValue().map().size()); } diff --git a/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientWatchersTest.java b/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientWatchersTest.java index 8eaa4721..e4c9aa3e 100644 --- a/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientWatchersTest.java +++ b/tarantool-core/src/test/java/io/tarantool/core/integration/IProtoClientWatchersTest.java @@ -58,8 +58,6 @@ private IProtoClient getClientAndConnect(TarantoolContainer tt) throws Except InetSocketAddress address = tt.mappedAddress(); IProtoClient client = new IProtoClientImpl(factory, factory.getTimerService()); client.connect(address, 3_000).get(); - // Complete the handshake with ping so that watchers are sent - client.ping().get(); return client; } @@ -179,17 +177,12 @@ private void testWatcherRecoveryAfterReconnectOnContainer( "box.broadcast('keyA', 'myEvent');" + "box.broadcast('keyB', {1, 2, 3});" + "box.broadcast('keyC', 'wontbecaught');"); - // let the first broadcast events arrive before the connection is closed Thread.sleep(100); client.close(); - Thread.sleep(200); + Thread.sleep(100); InetSocketAddress address = tt.mappedAddress(); client.connect(address, 3_000).get(); - // Re-trigger watcher registration after reconnect (same as getClientAndConnect) - client.ping().get(); - // Give time for handleClose to finish clearing stateContext on the Netty thread - // and for IPROTO_WATCH responses to arrive before sending new broadcast - Thread.sleep(1500); + Thread.sleep(1000); TarantoolContainerClientHelper.executeCommand( tt, From a14c5b208d4fce5412755e7bacb7ac67d105bea6 Mon Sep 17 00:00:00 2001 From: Dmitry Kasimovskiy Date: Fri, 25 Sep 2026 10:37:24 +0300 Subject: [PATCH 5/7] docs(changelog): describe error-driven watcher registration --- CHANGELOG.md | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 99201cbf..6f0002be 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,13 +8,13 @@ ### Bug fixes -- Register watchers (including the automatic `box.shutdown` watcher created when - `gracefulShutdown` is enabled) only after the session is authenticated, i.e. after - `authorize()` or `ping()` completes, instead of right after the greeting. Tarantool EE - builds with option `security.disable_guest: true` allow only `auth`, `ping`, `id` and - `vote` requests before authentication and answered the early `IPROTO_WATCH` with - `ER_AUTH_REQUIRED`; since 1.7.0 that watcher error failed the whole connect procedure - and put the connection into an endless reconnect loop. +- Handle `ER_AUTH_REQUIRED` for watcher registration. Watchers (including the automatic + `box.shutdown` watcher created when `gracefulShutdown` is enabled) are now deferred when + the server rejects their registration before authentication and re-registered after + `authorize()` completes. Tarantool EE builds with option `security.disable_guest: true` + reject pre-authentication watcher registration with `ER_AUTH_REQUIRED`; since 1.7.0 that + watcher error failed the whole connect procedure and put the connection into an endless + reconnect loop. ## [1.7.1] - 2026-08-31 From b0e13a268785907d3c6414759ca4f8c94f4a1a3c Mon Sep 17 00:00:00 2001 From: Dmitry Kasimovskiy Date: Fri, 25 Sep 2026 13:04:43 +0300 Subject: [PATCH 6/7] refactor(tarantool-core): move ER_AUTH_REQUIRED to IProtoConstant --- .../io/tarantool/core/protocol/fsm/WatcherStateMachine.java | 6 ++---- .../io/tarantool/core/protocol/requests/IProtoConstant.java | 1 + 2 files changed, 3 insertions(+), 4 deletions(-) diff --git a/tarantool-core/src/main/java/io/tarantool/core/protocol/fsm/WatcherStateMachine.java b/tarantool-core/src/main/java/io/tarantool/core/protocol/fsm/WatcherStateMachine.java index 8c889092..5c972fe2 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/protocol/fsm/WatcherStateMachine.java +++ b/tarantool-core/src/main/java/io/tarantool/core/protocol/fsm/WatcherStateMachine.java @@ -19,13 +19,11 @@ import io.tarantool.core.exceptions.ClientException; import io.tarantool.core.protocol.IProtoRequest; import io.tarantool.core.protocol.IProtoResponse; +import io.tarantool.core.protocol.requests.IProtoConstant; import io.tarantool.core.protocol.requests.IProtoWatch; public class WatcherStateMachine implements IProtoStateMachine { - /** Tarantool error code for requests rejected before authentication (EE only). */ - public static final int ER_AUTH_REQUIRED = 258; - private final Connection connection; private static final Logger log = LoggerFactory.getLogger(WatcherStateMachine.class); @@ -112,7 +110,7 @@ public boolean process(IProtoResponse message) { } else { ClientException error = new ClientException("watcher error: %s", message); registered.completeExceptionally(error); - if (onAuthRequired != null && message.getErrorCode() == ER_AUTH_REQUIRED) { + if (onAuthRequired != null && message.getErrorCode() == IProtoConstant.IPROTO_ERR_AUTH_REQUIRED) { log.debug("watcher '{}' rejected before authentication, deferred", key); onAuthRequired.accept(this); } else { diff --git a/tarantool-core/src/main/java/io/tarantool/core/protocol/requests/IProtoConstant.java b/tarantool-core/src/main/java/io/tarantool/core/protocol/requests/IProtoConstant.java index 24f4e4b0..3fa559f3 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/protocol/requests/IProtoConstant.java +++ b/tarantool-core/src/main/java/io/tarantool/core/protocol/requests/IProtoConstant.java @@ -18,6 +18,7 @@ public interface IProtoConstant { int IPROTO_ERROR_24 = 0x31; int IPROTO_ERROR_BASE = 0x8000; int IPROTO_ERR_ACCESS_DENIED = 0x2A; + int IPROTO_ERR_AUTH_REQUIRED = 0x102; int IPROTO_ERR_CREDS_MISMATCH = 0x2F; int IPROTO_ERR_INVALID_MSGPACK = 0x20; int IPROTO_ERR_NO_SUCH_PROC = 0x21; From fa5c74560efe80ae067b131542592312a6b3d56c Mon Sep 17 00:00:00 2001 From: Dmitry Kasimovskiy Date: Fri, 25 Sep 2026 13:06:51 +0300 Subject: [PATCH 7/7] style(tarantool-core): fix line length for spotless --- .../io/tarantool/core/protocol/fsm/WatcherStateMachine.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tarantool-core/src/main/java/io/tarantool/core/protocol/fsm/WatcherStateMachine.java b/tarantool-core/src/main/java/io/tarantool/core/protocol/fsm/WatcherStateMachine.java index 5c972fe2..566d64cc 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/protocol/fsm/WatcherStateMachine.java +++ b/tarantool-core/src/main/java/io/tarantool/core/protocol/fsm/WatcherStateMachine.java @@ -110,7 +110,8 @@ public boolean process(IProtoResponse message) { } else { ClientException error = new ClientException("watcher error: %s", message); registered.completeExceptionally(error); - if (onAuthRequired != null && message.getErrorCode() == IProtoConstant.IPROTO_ERR_AUTH_REQUIRED) { + if (onAuthRequired != null + && message.getErrorCode() == IProtoConstant.IPROTO_ERR_AUTH_REQUIRED) { log.debug("watcher '{}' rejected before authentication, deferred", key); onAuthRequired.accept(this); } else {