Infinite Subscriptions happening on client call

[vimaldermal]

We upgraded from micronaut 3.7.3 to 4.7.9.

public Single<HttpResponse<SampleResponse>> checkDirectCall(SampleRequest sampleRequest) { return sampleClientCall() .map(clientResponse-> HttpResponse.status(INTERNAL_SERVER_ERROR).body(new SampleResponse())); }

The sampleClientCall() client call is getting invoked multiple times sometimes. We were not able to reproduce issue locally. It happens when

we deploye the service to K8s.

When we set the status to OK, we are not facing the issue.

The same code is working as expected in 3.7.3

[vimaldermal]

@graemerocher @sdelamo could you please have a check and see if this has something to do with the upgrade ?

[TatyOko28]

Root Cause

1. Reactive Streams subscription behavior changed

In Micronaut 4.x, the HTTP client implementation changed from RxJava2 to Reactive Streams (Project Reactor behind the scenes, depending on your setup). The Single from sampleClientCall() may be cold and can be re-subscribed to multiple times if the stream is not properly cached.

2. Error handling in reactive chains

When you return an error status (INTERNAL_SERVER_ERROR), the reactive chain may treat this as a failure signal depending on how the HTTP client interprets non-2xx responses.

3. Kubernetes environment differences

In K8s, things that trigger the issue but not locally:

• Service mesh / sidecar proxies (Istio, Linkerd) may cause connection resets or retries
• Network latency can cause timeout-based retries
• Readiness/liveness probes may interrupt requests
• Different default timeouts in the K8s environment

4. Micronaut 4.x default retry behavior

Micronaut 4.x introduced more aggressive client retry defaults for certain error conditions (e.g., connection errors, certain HTTP status codes). What you're seeing might be an automatic retry on non-2xx responses.

Solutions

Solution 1: Cache the reactive stream (Recommended)

Use .cache() to ensure the client call is executed only once even if subscribed multiple times.

import io.reactivex.Single;
import io.micronaut.http.HttpResponse;
import io.micronaut.http.HttpStatus;

public Single<HttpResponse<SampleResponse>> checkDirectCall(SampleRequest sampleRequest) {
    return sampleClientCall()
        .cache()  // ← Prevents multiple subscriptions
        .map(clientResponse -> HttpResponse.status(HttpStatus.INTERNAL_SERVER_ERROR).body(new SampleResponse()));
}

Solution 2: Use .toObservable().firstOrError() instead of .cache()

public Single<HttpResponse<SampleResponse>> checkDirectCall(SampleRequest sampleRequest) {
    return sampleClientCall()
        .toObservable()
        .firstOrError()
        .map(clientResponse -> HttpResponse.status(HttpStatus.INTERNAL_SERVER_ERROR).body(new SampleResponse()));
}

Solution 3: Disable automatic retries on the HTTP client

Configure the HTTP client to not retry on error responses:

# application.yml
micronaut:
  http:
    client:
      retry-on-error: false  # Disables automatic retries

Or per client:

@Client(id = "sample-service", configuration = SampleClientConfiguration.class)
public interface SampleClient {
    // ...
}

import io.micronaut.context.annotation.ConfigurationProperties;
import io.micronaut.http.client.HttpClientConfiguration;

@ConfigurationProperties("sample-client")
public class SampleClientConfiguration extends HttpClientConfiguration {
    @Override
    public boolean getRetryOnError() {
        return false;  // Disable retries for this client
    }
}

Solution 4: Use Blocking or CompletableFuture instead of Single

If you don't strictly need reactive backpressure, switch to a non-reactive approach:

public HttpResponse<SampleResponse> checkDirectCall(SampleRequest sampleRequest) {
    SampleClientResponse clientResponse = sampleClientCallBlocking();  // or .block()
    return HttpResponse.status(HttpStatus.INTERNAL_SERVER_ERROR).body(new SampleResponse());
}

Or use CompletableFuture:

public CompletableFuture<HttpResponse<SampleResponse>> checkDirectCall(SampleRequest sampleRequest) {
    return sampleClientCallFuture()
        .thenApply(clientResponse -> 
            HttpResponse.status(HttpStatus.INTERNAL_SERVER_ERROR).body(new SampleResponse())
        );
}

Solution 5: Add error handling to prevent retry propagation

public Single<HttpResponse<SampleResponse>> checkDirectCall(SampleRequest sampleRequest) {
    return sampleClientCall()
        .onErrorReturn(throwable -> {
            // Convert error to a fallback response
            return new SampleClientResponse();  // or appropriate fallback
        })
        .map(clientResponse -> HttpResponse.status(HttpStatus.INTERNAL_SERVER_ERROR).body(new SampleResponse()));
}

Solution 6: Use Single.defer() with caching

public Single<HttpResponse<SampleResponse>> checkDirectCall(SampleRequest sampleRequest) {
    return Single.defer(() -> sampleClientCall())
        .cache()
        .map(clientResponse -> HttpResponse.status(HttpStatus.INTERNAL_SERVER_ERROR).body(new SampleResponse()));
}

Debugging: How to confirm multiple invocations

Add logging to see if the client call is being invoked multiple times:

private Single<SampleClientResponse> sampleClientCall() {
    log.info("sampleClientCall() invoked at {}", System.currentTimeMillis());
    // rest of the method
}

Check the logs in K8s. If you see the same timestamp (or timestamps within milliseconds) multiple times for the same request, you have re-subscription.

Environment-specific checks for K8s

Check if probes are causing duplicate requests

# In your deployment.yaml
livenessProbe:
  httpGet:
    path: /health
    port: 8080
  initialDelaySeconds: 30
  periodSeconds: 10

These probes might be hitting your endpoint unexpectedly. Add logging to track correlation IDs.

Check Istio/Service mesh retries

If using Istio, you may have destination rule retries:

apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: sample-service
spec:
  host: sample-service
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 1
    loadBalancer:
      simple: RANDOM
    outlierDetection:
      consecutiveErrors: 7
      interval: 30s
      baseEjectionTime: 30s
    # Retry configuration may be here:
    retry:
      attempts: 3
      perTryTimeout: 2s
      retryOn: 5xx

If retryOn: 5xx is present, your INTERNAL_SERVER_ERROR (500) will be retried.

Check Micronaut server retry configuration

micronaut:
  server:
    retry:
      enabled: true  # Default in some versions
      max-retries: 3
      retry-on: [INTERNAL_SERVER_ERROR, SERVICE_UNAVAILABLE]

Complete working solution for Micronaut 4.7.9

Here's a production-ready fix combining multiple safeguards:

import io.micronaut.http.HttpResponse;
import io.micronaut.http.HttpStatus;
import io.micronaut.http.annotation.Controller;
import io.micronaut.http.annotation.Post;
import io.reactivex.Single;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.atomic.AtomicLong;

@Controller("/api")
public class SampleController {

    private static final Logger log = LoggerFactory.getLogger(SampleController.class);
    private final SampleClient sampleClient;
    private final AtomicLong callCounter = new AtomicLong();

    public SampleController(SampleClient sampleClient) {
        this.sampleClient = sampleClient;
    }

    @Post("/check-direct-call")
    public Single<HttpResponse<SampleResponse>> checkDirectCall(SampleRequest sampleRequest) {
        long invocationId = callCounter.incrementAndGet();
        log.info("Request {} - checkDirectCall invoked", invocationId);

        return Single.defer(() -> {
                log.info("Request {} - executing sampleClientCall()", invocationId);
                return sampleClient.call(sampleRequest);
            })
            .cache()  // Critical: prevents multiple subscriptions
            .map(clientResponse -> {
                log.info("Request {} - mapping to error response", invocationId);
                return HttpResponse.<SampleResponse>status(HttpStatus.INTERNAL_SERVER_ERROR)
                    .body(new SampleResponse());
            })
            .doOnError(throwable -> {
                log.error("Request {} - error occurred", invocationId, throwable);
            });
    }
}

With application.yml safeguards:

micronaut:
  http:
    client:
      retry-on-error: false  # Disable automatic retries
  server:
    retry:
      enabled: false  # Disable server-side retries
    netty:
      max-initial-line-length: 4096
      max-header-size: 8192