Force Reconnect API #1100

scottf · 2024-03-20T16:45:02Z

Implement the Force Reconnect API (Add reconnect method nats-architecture-and-design#259)
Better handling socket write timeout is triggered, uses force reconnect

scottf · 2024-03-20T20:48:11Z

src/examples/java/io/nats/examples/chaosTestApp/OutputErrorListener.java

+    @Override
+    public void socketWriteTimeout(Connection conn) {
+        Output.controlMessage(this.id, supplyMessage("SEVERE socketWriteTimeout", conn, null, null));
+    }


This was added since the class is an ErrorListener and socketWriteTimeout was added.

scottf · 2024-03-20T20:49:50Z

src/examples/java/io/nats/examples/jetstream/ResilientPublisher.java

+//                .delay(1)
+//                .reportFrequency(1000);
+// END CORE PUBLISHER EXAMPLE
+


This is an example class that demonstrates resilient publishing. I added some functionality to it, for instance it only demoed Jet Stream publishing, but now can support core publishing. There are also some other improvements.

scottf · 2024-03-20T20:50:29Z

src/examples/java/io/nats/examples/jetstream/simple/IterableConsumerExample.java

@@ -57,7 +57,7 @@ public static void main(String[] args) {
            }

            System.out.println("Starting publish...");
-            ResilientPublisher publisher = ResilientPublisher.newInstanceQuietAndJitter(jsm, STREAM, SUBJECT, MESSAGE_PREFIX, 10);
+            ResilientPublisher publisher = new ResilientPublisher(nc, jsm, STREAM, SUBJECT).basicDataPrefix(MESSAGE_PREFIX).jitter(10);


This was changed because ResilentPublisher was changed and it uses that class.

scottf · 2024-03-20T20:50:38Z

src/examples/java/io/nats/examples/jetstream/simple/MessageConsumerExample.java

@@ -43,7 +43,7 @@ public static void main(String[] args) {
            createOrReplaceStream(jsm, STREAM, SUBJECT);

            System.out.println("Starting publish...");
-            ResilientPublisher publisher = ResilientPublisher.newInstanceQuietAndJitter(jsm, STREAM, SUBJECT, MESSAGE_PREFIX, 10);
+            ResilientPublisher publisher = new ResilientPublisher(nc, jsm, STREAM, SUBJECT).basicDataPrefix(MESSAGE_PREFIX).jitter(10);


This was changed because ResilentPublisher was changed and it uses that class.

scottf · 2024-03-20T20:50:55Z

src/main/java/io/nats/client/Connection.java

+     * @throws IOException
+     * @throws InterruptedException
+     */
+    void forceReconnect() throws IOException, InterruptedException;


This is the new api in the Connection object.

scottf · 2024-03-20T20:52:12Z

src/main/java/io/nats/client/ConnectionListener.java

        /** Server Sent a lame duck mode. */
-        LAME_DUCK("nats: lame duck mode");
+        LAME_DUCK(false, "lame duck mode");


I modified this class for more granular information about the event and to allow for different types of reporting. It is backward compatible, meaning that the toString did not change text.

scottf · 2024-03-20T20:53:13Z

src/main/java/io/nats/client/ErrorListener.java

+     *
+     * @param conn The connection that had the issue
+     */
+    default void socketWriteTimeout(Connection conn) {}


This is the new ErrorListener event. It has a default implementation to be backward compatible.

scottf · 2024-03-20T20:54:17Z

src/main/java/io/nats/client/ErrorListener.java

+     *              to the message like ", foo: &lt;fooValue&gt;, bar-&lt;barValue&gt;".
+     * @return
+     */
+    default String supplyMessage(String label, Connection conn, Consumer consumer, Subscription sub, Object... pairs) {


This function used to be part of the ErrorListenerLoggerImpl class, but it was moved down into the interface since so it can be more easily used in ErrorListener implementations.

scottf · 2024-03-20T20:55:25Z

src/main/java/io/nats/client/impl/ErrorListenerConsoleImpl.java

+import io.nats.client.*;
+import io.nats.client.support.Status;
+
+public class ErrorListenerConsoleImpl implements ErrorListener {


The console implementation is less verbose than the logger impl, nothing really exciting.

scottf · 2024-03-20T20:55:53Z

src/main/java/io/nats/client/impl/ErrorListenerLoggerImpl.java

@@ -23,30 +22,6 @@ public class ErrorListenerLoggerImpl implements ErrorListener {

    private final static Logger LOGGER = Logger.getLogger(ErrorListenerLoggerImpl.class.getName());

-    private String supplyMessage(String label, Connection conn, Consumer consumer, Subscription sub, Object... pairs) {


As already noted, this was moved down into the interface.

scottf · 2024-03-20T20:56:26Z

src/main/java/io/nats/client/impl/ErrorListenerLoggerImpl.java

+    @Override
+    public void socketWriteTimeout(Connection conn) {
+        LOGGER.severe(() -> supplyMessage("socketWriteTimeout", conn, null, null));
+    }


Implementation specific implementation since the interface has a no-op default implementation.

scottf · 2024-03-20T20:58:10Z

src/main/java/io/nats/client/impl/MessageQueue.java

+        this.requestCleanupInterval = requestCleanupInterval;
+    }
+
+    MessageQueue(MessageQueue source) {


This allows a message queue to be created from another message queue. It is used when a writer is failing because it's blocked or a force reconnect, to take the write queue from the writer that is being discard and moving all the items (messages) in the queue to the queue in the new writer so they are not lost.

scottf · 2024-03-20T20:58:56Z

src/main/java/io/nats/client/impl/MessageQueue.java

@@ -33,11 +33,13 @@ class MessageQueue {
    protected final AtomicLong length;
    protected final AtomicLong sizeInBytes;
    protected final AtomicInteger running;
-    protected final boolean singleThreadedReader;
+    protected final boolean singleReaderMode;


made the variable name more consistent with it's usage and to match it's getter.

scottf · 2024-03-20T20:59:45Z

src/main/java/io/nats/client/impl/MessageQueue.java

    protected final LinkedBlockingQueue<NatsMessage> queue;
    protected final Lock filterLock;
+    protected final int publishHighwaterMark;


state needed for when constructing a queue from a queue.

scottf · 2024-03-20T20:59:49Z

src/main/java/io/nats/client/impl/MessageQueue.java

    protected final boolean discardWhenFull;
    protected final long offerTimeoutMillis;
+    protected final Duration requestCleanupInterval;


state needed for when constructing a queue from a queue.

scottf · 2024-03-20T21:00:29Z

src/main/java/io/nats/client/impl/NatsConnection.java

-    private final NatsConnectionReader reader;
-    private final NatsConnectionWriter writer;
+    private NatsConnectionReader reader;
+    private NatsConnectionWriter writer;


these are no longer final, they will be swapped out on a force reconnect.

scottf · 2024-03-20T21:02:42Z

src/main/java/io/nats/client/impl/NatsConnection.java

-            this.writer.stop().get(timeoutNanos, TimeUnit.NANOSECONDS);
+            if (writer.isRunning()) {
+                this.writer.stop().get(timeoutNanos, TimeUnit.NANOSECONDS);
+            }


sometimes stop times out even when it's not running. This guard prevents that since it's already been stopped.

scottf · 2024-03-20T21:03:48Z

src/main/java/io/nats/client/impl/NatsConnectionReader.java

-                dataPort.shutdownInput();
-            } catch (IOException e) {
-                // we don't care, we are shutting down anyway
+        if (running.get()) {


If it's not running, this has already been called, no need to go into the logic.

scottf · 2024-03-20T21:06:23Z

src/main/java/io/nats/client/impl/NatsConnectionReader.java

+            // if already not running, an IOE is not unreasonable in a transition state
+            if (running.get()) {
+                this.connection.handleCommunicationIssue(io);
+            }


if already not running, an IOE is not unreasonable in a transition state

scottf · 2024-03-20T21:07:19Z

src/main/java/io/nats/client/impl/NatsConnectionWriter.java

@@ -75,6 +75,27 @@ class NatsConnectionWriter implements Runnable {
        reconnectBufferSize = options.getReconnectBufferSize();
    }

+    NatsConnectionWriter(NatsConnectionWriter sourceWriter) {


Added to replace an existing writer but keeps it's relevant state including the message queue.

scottf · 2024-03-20T21:07:37Z

src/main/java/io/nats/client/impl/NatsConnectionWriter.java

-            this.outgoing.filter((msg) ->
-                msg.isProtocol() &&
-                    (msg.protocolBab.equals(OP_PING_BYTES) || msg.protocolBab.equals(OP_PONG_BYTES)));
+        if (running.get()) {


if it's not running, don't need to go into the stop logic.

scottf · 2024-03-20T21:08:42Z

src/main/java/io/nats/client/impl/SocketDataPortWithWriteTimeout.java

@@ -36,12 +36,16 @@ class WriteWatcherTask extends TimerTask {
        public void run() {
            //  if now is after when it was supposed to be done by
            if (System.nanoTime() > writeMustBeDoneBy) {
+                writeWatcherTimer.cancel(); // we don't need to repeat this
+                connection.executeCallback((c, el) -> el.socketWriteTimeout(c));


This is the callback to the new ErrorListener method.

scottf · 2024-03-20T21:09:43Z

src/main/java/io/nats/client/impl/SocketDataPortWithWriteTimeout.java

                }
-                catch (InterruptedException e) {
-                    // do nothing
+                catch (IOException ignore) {}


try closing the output. Most likely going to fail since it was blocked, hence the catch, but also could have actually be freed up in the middle, so close it anyway, it's state is bad.

MauriceVanVeen

Thank you for placing the comments throughout, it helped a lot!

Pretty much only have one comment, but am happy to approve afterward.

MauriceVanVeen · 2024-03-21T12:45:30Z

src/main/java/io/nats/client/impl/NatsConnection.java

+            updateStatus(Status.DISCONNECTED);
+            reader.stop();
+            writer.stop();
+            writer = new NatsConnectionWriter(writer);


This comment mentions making both reader and writer non-final to allow swapping them on forceReconnect.
https://github.com/nats-io/nats.java/pull/1100/files#r1532862943

Is it correct that only the writer is overwritten here, and not reader? Or is that done as part of the reconnect() call?

Yeah, at one point I was going to swap both the reader and writer. I'll change the reader back, I can also swap it in the future if I needed.

MauriceVanVeen · 2024-03-21T12:55:08Z

src/test/java/io/nats/client/impl/SocketDataPortBlockSimulator.java

+    }
+
+    @Override
+    public void afterContruct(Options options) {


Just wondering, the interface was not changed, but could this be a small typo?
afterConstruct

Yikes. This was already released as public but I'm fixing it anyway as a bug since I'm 99.9999% sure no one is even aware of it, it's essentially internal

scottf · 2024-03-21T19:07:35Z

Thank you for placing the comments throughout, it helped a lot!
I do this as part of my own review like rubber ducking, plus it makes it easier for anyone else to know what I'm doing since I never know who might review and what their primary coding language is.

MauriceVanVeen

LGTM, nice work!

Force Reconnect API and better socket write timeout handling

9a70651

scottf requested a review from MauriceVanVeen March 20, 2024 16:45

scottf changed the title ~~Force Reconnect API and better socket write timeout handling~~ Force Reconnect API Mar 20, 2024

scottf mentioned this pull request Mar 20, 2024

Add reconnect method nats-io/nats-architecture-and-design#259

Open

13 tasks

scottf commented Mar 20, 2024

View reviewed changes

self review

ce6c87a

MauriceVanVeen reviewed Mar 21, 2024

View reviewed changes

Address review comments.

0dc5be2

scottf requested a review from MauriceVanVeen March 21, 2024 19:07

MauriceVanVeen approved these changes Mar 22, 2024

View reviewed changes

scottf merged commit 8a69711 into main Mar 22, 2024
2 checks passed

scottf deleted the force-reconnect branch March 22, 2024 17:30

scottf mentioned this pull request Mar 26, 2024

Add reconnect method #1085

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Force Reconnect API #1100

Force Reconnect API #1100

scottf commented Mar 20, 2024 •

edited

Loading

scottf Mar 20, 2024

scottf Mar 20, 2024

scottf Mar 20, 2024

scottf Mar 20, 2024

scottf Mar 20, 2024

scottf Mar 20, 2024

scottf Mar 20, 2024 •

edited

Loading

scottf Mar 20, 2024

scottf Mar 20, 2024

scottf Mar 20, 2024

scottf Mar 20, 2024

scottf Mar 20, 2024

scottf Mar 20, 2024

scottf Mar 20, 2024

scottf Mar 20, 2024

scottf Mar 20, 2024

scottf Mar 20, 2024

scottf Mar 20, 2024

scottf Mar 20, 2024

scottf Mar 20, 2024

scottf Mar 20, 2024

scottf Mar 20, 2024

scottf Mar 20, 2024

MauriceVanVeen left a comment

MauriceVanVeen Mar 21, 2024

scottf Mar 21, 2024

MauriceVanVeen Mar 21, 2024

scottf Mar 21, 2024 •

edited

Loading

scottf commented Mar 21, 2024

MauriceVanVeen left a comment

		@@ -23,30 +22,6 @@ public class ErrorListenerLoggerImpl implements ErrorListener {

		private final static Logger LOGGER = Logger.getLogger(ErrorListenerLoggerImpl.class.getName());

		private String supplyMessage(String label, Connection conn, Consumer consumer, Subscription sub, Object... pairs) {

Force Reconnect API #1100

Force Reconnect API #1100

Conversation

scottf commented Mar 20, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

scottf Mar 20, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

MauriceVanVeen left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

scottf Mar 21, 2024 • edited Loading

Choose a reason for hiding this comment

scottf commented Mar 21, 2024

MauriceVanVeen left a comment

Choose a reason for hiding this comment

scottf commented Mar 20, 2024 •

edited

Loading

scottf Mar 20, 2024 •

edited

Loading

scottf Mar 21, 2024 •

edited

Loading