Reconnection

Overview

Production connections drop for reasons that have nothing to do with your application code: the broker restarts during a rolling upgrade, a load balancer fails over, a network blip severs the gRPC stream. Without built-in reconnection, every client in your fleet needs its own hand-rolled retry-connect loop, and it's easy to get the backoff wrong — too aggressive and you hammer a recovering broker, too slow and you leave the application dark longer than necessary. Automatic reconnection moves that logic into the client itself, so the connection self-heals without any code the application has to write or maintain.

It works by setting reconnectIntervalSeconds(...) on PubSubClient.builder() — the SDK doubles this interval on each failed attempt up to a default maximum. When the broker becomes unreachable, the subscription's onErrorCallback fires and the SDK automatically re-establishes the stream and resumes delivery once connectivity returns, re-registering the subscription without any application code. Gotchas: the error callback fires whenever the underlying stream breaks, not just on permanent failures, so treat it as a signal rather than a fatal event; publishEvent() calls made during the outage window throw a KubeMQException immediately — the interval governs the subscription's reconnection, not individual publishes, so wrap sends in your own try/catch and retry if you need guaranteed delivery during outages; and there's no attempt cap in this policy, so a permanently dead broker will be retried forever unless you add your own circuit breaker.

Prerequisites

• KubeMQ server running on localhost:50000
• Java SDK installed (implementation 'io.kubemq.sdk:kubemq-sdk-Java:3.1.1' (Gradle) or Maven dependency from Getting Started)

Code

package io.kubemq.example.errorhandling;

import io.kubemq.sdk.common.ServerInfo;
import io.kubemq.sdk.pubsub.*;

/**
 * Reconnection Example
 *
 * Demonstrates built-in reconnection handling for KubeMQ subscriptions.
 */
public class ReconnectionExample {

    private static final String ADDRESS = "localhost:50000";
    private static final String CLIENT_ID = "java-errorhandling-reconnection-client";

    public static void main(String[] args) throws InterruptedException {
        System.out.println("=== Reconnection Handling ===\n");

        // Create a client with reconnection enabled
        PubSubClient client = PubSubClient.builder()
                .address(ADDRESS)
                .clientId(CLIENT_ID)
                .reconnectIntervalSeconds(1)
                .build();

        client.ping();
        // Create channel for the subscription
        client.createEventsChannel("java-errorhandling.reconnect-test");

        // Subscribe with error callback (SDK auto-reconnects on errors)
        EventsSubscription subscription = EventsSubscription.builder()
                .channel("java-errorhandling.reconnect-test")
                .onReceiveEventCallback(event ->
                        System.out.println("  Received: " + new String(event.getBody())))
                .onErrorCallback(error -> {
                    System.out.println("  [ERROR] " + error);
                    System.out.println("  (SDK will attempt reconnection automatically)");
                })
                .build();

        // Start the subscription
        client.subscribeToEvents(subscription);
        System.out.println("Subscription active with automatic reconnection.\n");

        // Send a test message
        client.publishEvent(EventMessage.builder()
                .channel("java-errorhandling.reconnect-test")
                .body("Test message".getBytes()).build());

        Thread.sleep(500);

        // Clean up resources
        subscription.cancel();
        client.deleteEventsChannel("java-errorhandling.reconnect-test");
        client.close();
        System.out.println("Reconnection example completed.");
    }
}

How It Works

• reconnectIntervalSeconds(1) sets the base interval for the SDK's built-in exponential backoff loop; on each failed reconnection attempt the interval doubles until it reaches the maximum (default 60 s).
• When the broker is temporarily unavailable, the onErrorCallback on the subscription fires; the SDK then automatically attempts to re-establish the gRPC stream and resume delivery once connectivity is restored.
• The subscription is re-registered automatically after reconnection — no application code is needed to re-subscribe.
• publishEvent() during a disconnection window throws a KubeMQException; wrap sends in try/catch and apply your own retry logic if the application requires guaranteed delivery during outages.

Related

• Java SDK Reference
• Connection Error
• Graceful Shutdown