SpringBoot Custom Actuator Endpoint with MediaType as text/event-stream is not working

@bclozel As per your suggestion, I tried to utilize the most recent supported version of Spring and was still able to replicate the problem. Below the are the details

Versions That i have used Java:- OpenJDK-17 Spring boot: 3.2.10 reactor-core: 3.6.10

Depedecies implementation("org.springframework.boot:spring-boot-starter") implementation("org.springframework.boot:spring-boot-starter-webflux") implementation("org.springframework.boot:spring-boot-starter-actuator") implementation("org.springframework.boot:spring-boot-starter-aop")

    implementation("io.github.resilience4j:resilience4j-spring-boot3")
implementation("io.github.resilience4j:resilience4j-all") // Optional, only required when you want to use the Decorators class
implementation("io.github.resilience4j:resilience4j-reactor")
implementation("io.micrometer:micrometer-registry-prometheus")

//implementation("de.codecentric:chaos-monkey-spring-boot:2.6.1")

implementation("io.vavr:vavr-jackson:0.10.3")
testImplementation("org.springframework.boot:spring-boot-starter-test")
testImplementation("io.projectreactor:reactor-test")

Reproducing steps Try to copy the two rest and actuator controllers, add the reactor and resilience dependencies of above if you haven't already, and then run your application.

Problem: I created a custom actuator endpoint to expose a data stream, however despite events being triggered, requests are taking a long time to complete and no data is being released in the end.

This is my simple controller package io.github.robwin.controller;

import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry; import io.github.resilience4j.circuitbreaker.event.CircuitBreakerEvent; import io.github.resilience4j.common.circuitbreaker.monitoring.endpoint.CircuitBreakerEventDTOFactory; import io.vavr.collection.List; import io.vavr.collection.Seq; import org.springframework.boot.actuate.endpoint.annotation.Endpoint; import org.springframework.boot.actuate.endpoint.annotation.ReadOperation; import org.springframework.http.codec.ServerSentEvent; import org.springframework.stereotype.Component; import reactor.core.publisher.Flux;

import java.time.Duration; import java.util.function.Function;

import static io.github.resilience4j.reactor.adapter.ReactorAdapter.toFlux;

@Component @Endpoint(id = "test-sse") public class TestSseController {

private final CircuitBreakerRegistry circuitBreakerRegistry;

private final ObjectMapper jsonMapper = new ObjectMapper();

public TestSseController(
        CircuitBreakerRegistry circuitBreakerRegistry) {
    this.circuitBreakerRegistry = circuitBreakerRegistry;
}

@ReadOperation(produces = "text/event-stream")
public Flux<ServerSentEvent<String>> streamEvents() {
    return Flux.interval(Duration.ofSeconds(1))
            .map(sequence -> ServerSentEvent.<String> builder()
                    .id(String.valueOf(sequence))
                    .event("pingpong")
                    .data("ping")
                    .build());
}

//@ReadOperation(produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<String>> getAllCircuitBreakerServerSid() {
    Seq<Flux<CircuitBreakerEvent>> eventStreams = List.ofAll(circuitBreakerRegistry.getAllCircuitBreakers().stream()
            .map(circuitBreaker -> toFlux(circuitBreaker.getEventPublisher())));
    Function<CircuitBreakerEvent, String> data = getCircuitBreakerEventStringFunction();
    return Flux.merge(eventStreams).map(
            cbEvent -> ServerSentEvent.<String>builder()
                    .id(cbEvent.getCircuitBreakerName())
                    .event(cbEvent.getEventType().name())
                    .data(data.apply(cbEvent))
                    .build()
    );
}

private Function<CircuitBreakerEvent, String> getCircuitBreakerEventStringFunction() {
    return cbEvent -> {
        try {
            return jsonMapper.writeValueAsString(
                    CircuitBreakerEventDTOFactory.createCircuitBreakerEventDTO(cbEvent)
            );
        } catch (JsonProcessingException e) {
            /* ignore silently */
        }
        return "";
    };
}

}

Working fine with the RestController I noticed that, while the same configuration functions perfectly when using the restController, it does not function when using the endpoint and readOperation annotation. package io.github.robwin.controller;

import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry; import io.github.resilience4j.circuitbreaker.event.CircuitBreakerEvent; import io.github.resilience4j.common.circuitbreaker.monitoring.endpoint.CircuitBreakerEventDTOFactory; import io.vavr.collection.List; import io.vavr.collection.Seq; import org.springframework.boot.actuate.endpoint.annotation.Selector; import org.springframework.http.MediaType; import org.springframework.http.codec.ServerSentEvent; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RestController; import reactor.core.publisher.Flux;

import java.time.Duration; import java.util.function.Function;

import static io.github.resilience4j.reactor.adapter.ReactorAdapter.toFlux;

@RestController public class TestSseControllerRest {

@GetMapping(value = "/test-rest-name", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
//@ReadOperation(produces = "text/event-stream" )
public Flux<ServerSentEvent<String>> testSse(@Selector String name) {
    return Flux.interval(Duration.ofSeconds(1))
               .map(seq -> ServerSentEvent.builder("Event #" + seq).build());
}

private final CircuitBreakerRegistry circuitBreakerRegistry;

private final ObjectMapper jsonMapper = new ObjectMapper();

public TestSseControllerRest(
        CircuitBreakerRegistry circuitBreakerRegistry) {
    this.circuitBreakerRegistry = circuitBreakerRegistry;
}


@GetMapping(value = "/test-rest", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<String>> getAllCircuitBreakerServerSid() {
    Seq<Flux<CircuitBreakerEvent>> eventStreams = List.ofAll(circuitBreakerRegistry.getAllCircuitBreakers().stream()
            .map(circuitBreaker -> toFlux(circuitBreaker.getEventPublisher())));
    Function<CircuitBreakerEvent, String> data = getCircuitBreakerEventStringFunction();
    return Flux.merge(eventStreams).map(
            cbEvent -> ServerSentEvent.<String>builder()
                    .id(cbEvent.getCircuitBreakerName())
                    .event(cbEvent.getEventType().name())
                    .data(data.apply(cbEvent))
                    .build()
    );
}

private Function<CircuitBreakerEvent, String> getCircuitBreakerEventStringFunction() {
    return cbEvent -> {
        try {
            return jsonMapper.writeValueAsString(
                    CircuitBreakerEventDTOFactory.createCircuitBreakerEventDTO(cbEvent)
            );
        } catch (JsonProcessingException e) {
            /* ignore silently */
        }
        return "";
    };
}

}

Below are screenshots and images of instances that are both functioning and not working.

I could see that the event data was being generated when I attempted to use the controller.

However, it appears that no data is being updated for the actuator endpoint; though, all events between the controller and actuator endpoint are the same.

Comment From: wilkinsona

@talasila-sairam please turn the code snippets above into something minimal that we can easily run. To be minimal, it should not depend on resilience4j unless it's impossible to reproduce the problem without it. Once you have such a sample, please comment on https://github.com/spring-projects/spring-boot/issues/42470 and we can re-open the issue if necessary. You can share the sample with us by pushing it to a separate repository on GitHub or by zipping it up and attaching it to #42470.