question about nats cluster failover

[juanxiuxu]

We have a Java application that consumes messages from NATS using the code below with the doConsume=true flag enabled.

public void connect() {
      String logme = " :: Connecting to server: " + url_socket;
      processLogWrap(logme);

      try {
         MyErrorListener el = new MyErrorListener();
         MyConnectionListener cl = new MyConnectionListener();

         Options.Builder builder = new Options.Builder()
         .server(url_socket)
         //.connectionTimeout(Duration.ofSeconds(10))
         //.pingInterval(Duration.ofSeconds(10))
         .reconnectWait(Duration.ofSeconds(nats_reconnect_wait))
         //.reconnectBufferSize(200 * 1024 * 1024)
         .maxReconnects(nats_max_reconnects)  // Enable native reconnect for network failures
         //.verbose()
         .connectionListener(cl)
         .errorListener(el);

         // Add username and password authentication if provided
         if (!nats_username.isEmpty()) {
            builder = builder.userInfo(nats_username, nats_password);
            logme = " :: Using authentication for user: " + nats_username + " password: " + nats_password;
            processLogWrap(logme);
         }

         if (tls_secure) {
            // Create a trust manager that trusts all certificates
            TrustManager[] trustAllCerts = new TrustManager[] {
               new X509TrustManager() {
                  public X509Certificate[] getAcceptedIssuers() {
                     return new X509Certificate[0];
                  }
                  public void checkClientTrusted(X509Certificate[] certs, String authType) {
                  }
                  public void checkServerTrusted(X509Certificate[] certs, String authType) {
                  }
               }
            };
            
            // Create SSL context with the trust-all manager
            SSLContext sslContext = SSLContext.getInstance("TLS");
            sslContext.init(null, trustAllCerts, new SecureRandom());
            
            builder = builder.secure().sslContext(sslContext);
            
            logme = " :: TLS enabled with SSL certificate verification bypassed";
            processLogWrap(logme);
         }

         m_connection = Nats.connect(builder.build());

         /*JetStreamManagement jsm = nc.jetStreamManagement();
         StreamConfiguration sc = StreamConfiguration.builder()
                                  .name(nats_stream)
                                  //.storageType(StorageType.File)
                                  .subjects(nats_subject)
                                  .build();
         StreamInfo streamInfo = jsm.addStream(sc);
         JsonUtils.printFormatted(streamInfo);
         */
         m_js = m_connection.jetStream();

         //setjavaNatsUp();

         //if you are a consumer, create Dispatcher here
         if (doConsume) {
            logme = " :: Subscribing to stream: " + nats_stream + " subject: " + nats_subject + " durable:" + nats_durable;
            processLogWrap(logme);

            ConsumerConfiguration consumerConfig = ConsumerConfiguration.builder()
                                                   .deliverPolicy(DeliverPolicy.New)
                                                   .ackWait(Duration.ofSeconds(2))
                                                   .build();

            PushSubscribeOptions options = PushSubscribeOptions.builder()
                                           .stream(nats_stream)
                                           .durable(nats_durable)
                                           .configuration(consumerConfig)
                                           .build();

            if (test_toggle) {
               /// This async style is easier to implement but not working with MDE current sync setup
               m_dispatcher = m_connection.createDispatcher(); //for consuming
               m_sub = m_js.subscribe(nats_subject, m_dispatcher, msg -> postToRecvQueueWrap(msg), true, options);
               m_connection.flush(Duration.ofSeconds(5));

            } else {
               m_sub = m_js.subscribe(nats_subject, nats_durable, options);
               m_connection.flush(Duration.ofSeconds(5));
             
               m_listenerThread = new Thread(new Runnable() {
                  @Override
                  public void run() {
                     startlistener();
                  }
               });
               m_listenerThread.start();
            }
         }

         logme = " :: Connection Success";
         processLogWrap(logme);

         running = true;

      } catch (Exception e) {
         logme = " :: Error: Exception while connecting to nats";
         processLogWrap(logme);

         e.printStackTrace();
         running = false;
      }
   }

The application monitors whether the NATS cluster fails over from AWS use1 region to euw1 region through a monitoring thread.
In the failover scenario:
The application can still connect to the use1 cluster and consume data
When our backend service encounters issues publishing to NATS, the application forces a new connection to euw1 by calling the NATS API's forceReconnect() method
the monitor thread code such as below:

private void monitorIpChange() {
     try {
        // 1. Check connection status
        if (!isConnected()) {
           processLogWrap(" :: NATS connection is not in CONNECTED state, skipping IP change check.");
           return;
        }
        // 2. Check if IPs have changed
        if (checkIpsChanged()) {
           try {
              if (isConnected() && running) {
                 processLogWrap(" :: [IPCheck] Calling forceReconnect due to IP/DNS change");
                 m_connection.forceReconnect();
                 m_needForceReconnect.set(true);
                 processLogWrap(" :: [IPCheck] Set m_needForceReconnect  = true");
              }
           } catch (Exception e) {
              processLogWrap(" :: Error during forceReconnect: " + e.getMessage());
           }
        }
     } catch (Exception e) {
        processLogWrap(" :: Error during IP change check: " + e.getMessage());
     }
  }

Before calling forceReconnect(), do we need to explicitly call unsubscribe() to clean up the old use1 subscription?

Since NATS push consumer load-balances message distribution among all subscriptions, if during the failover migration period, the old use1 subscription and the new euw1 migration subscription coexist (using the same durable consumer name), messages will be scattered between the two subscriptions, causing the migration subscription to fail to receive all messages.

[scottf]

Before calling forceReconnect(), do we need to explicitly call unsubscribe() to clean up the old use1 subscription?

No. Subscriptions will be re-made. Can you experiment to see what happens? What are your results?

I don't understand the monitorIpChange code. If the connection detects a disconnection, it will automatically reconnect, unless you have turned that ability off in the Options.

You should look into simplified consuming instead of push based consuming. https://github.com/nats-io/nats.java/tree/main/src/examples/java/io/nats/examples/jetstream/simple

[juanxiuxu]

Thanks @scottf.

let me explain more.

for my failover scenario, our application can connect to use1 nats cluster successfully but our backend service which is responsible for publishing data for the consumer application doesn't work. in this case, our application will do a failover by monitoring Route53 mapping for nats cluster to check whether need forceReconnect to switch to euw1 nats cluster. As we still can connect to use1 nats cluster when failover, it won't do automatically reconnect unless we call forceReconnect.

As I mentioned before, Since NATS push consumer load-balances message distribution among all subscriptions, if during the failover migration period, the old use1 subscription and the new use1 migration subscription which is responsible for moving the unprocessed data to euw1 coexist (using the same durable consumer name), messages will be scattered between the two subscriptions, causing the migration subscription to fail to receive all messages.

Could you let me know based on current implementation, how we can make sure the migration subscription to receive all unprocessed message? Do we need explicitly call unsubscribe before forceReconnect?

please let me know if it is clear to you.

Could you share some lights on this?

Thanks in advance.

[scottf]

1. Pull consumers automatically balance the same as push queue subscriptions without any extra work. Simplification uses pull.
2. Any reconnect, including force reconnect, subscriptions are re-established

Reconnects go to the same cluster they were already connected too, so I'm not sure what a force reconnect buys you, unless you provided your own server pool implementation to force connecting to another cluster.

If you do reconnect to another cluster, and the stream is not mirrored or the cluster is not part of a super cluster, subscription won't matter as the stream won't exist.

I still don't understand why you need to reconnect if there is no connection loss. When the publishers comes back on line, the consumer will start getting messages again. And if it has consumed less messages than are already in the stream if publishing is very fast, then this gives consumers time to catch up.

In order to ensure processing of all messages in a stream for a given consumer, you must use the ack feature. If a message is not acked within the ack wait timeframe, it will be redelivered.

[juanxiuxu]

Thanks @scottf