Server-Sent Events (SSE) provide a simple and efficient way for servers to push real-time updates to clients over HTTP with long lived connections. When deploying a Spring Boot application with SSE on Kubernetes, there are several challenges to address, especially when dealing with multiple replicas behind a LoadBalancer.
In this article, I will cover:
- How SSE works in a standard Spring Boot application.
- Why a naive approach fails in a Kubernetes setup.
- How to manage scalability using Redis Pub/Sub.
- Deploying the solution on Minikube with a LoadBalancer.
A More Naive Approach: Broadcasting to All Clients
A very basic approach to implementing SSE in Spring Boot would be to simply create an SseEmitter and broadcast messages to all clients connected to the endpoint. Below is an example:
@RestController
@RequestMapping("/sse")
public class SseBroadcastController {
private final List<SseEmitter> emitters = new CopyOnWriteArrayList<>();
@GetMapping("/subscribe")
public SseEmitter subscribe() {
SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);
emitters.add(emitter);
emitter.onCompletion(() -> emitters.remove(emitter));
emitter.onTimeout(() -> emitters.remove(emitter));
return emitter;
}
@PostMapping("/publish")
public ResponseEntity<String> publish(@RequestParam String message) {
for (SseEmitter emitter : emitters) {
try {
emitter.send(SseEmitter.event().data(message));
} catch (IOException e) {
emitter.complete();
}
}
return ResponseEntity.ok("Message sent");
}
}
Why This is Problematic
- Security Risk: Any client subscribing to the SSE endpoint would receive all messages, even if they were not intended recipients.
- Lack of Client Identification: There’s no way to target messages to specific users, making the approach impractical for personalized updates.
- Scalability Issues: This setup doesn’t scale well in a multi-instance Kubernetes deployment, as each instance maintains its own list of emitters without synchronization.
Understanding SSE in Spring Boot
A more controlled approach would involve identifying clients and storing SseEmitter instances based on client IDs. Here, clientID can be anything that is user-specific, like JWT’s.
SSE endpoints don’t allow headers to be passed so it is always a good idea to pass your clientID via Request Parameters or Path Variables.
Here’s a simple setup:
@RestController
@RequestMapping("/sse")
public class SseController {
private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();
@GetMapping("/subscribe/{clientId}")
public SseEmitter subscribe(@PathVariable String clientId) {
SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);
emitters.put(clientId, emitter);
emitter.onCompletion(() -> emitters.remove(clientId));
emitter.onTimeout(() -> emitters.remove(clientId));
return emitter;
}
@PostMapping("/publish")
public ResponseEntity<String> publish(@RequestParam String clientId, @RequestParam String message) {
SseEmitter emitter = emitters.get(clientId);
if (emitter != null) {
try {
emitter.send(SseEmitter.event().data(message));
} catch (IOException e) {
emitter.complete();
emitters.remove(clientId); // Remove emitter if sending fails
}
return ResponseEntity.ok("Message sent to client: " + clientId);
} else {
return ResponseEntity.status(HttpStatus.NOT_FOUND).body("Client not connected");
}
}
}
Issues with this implementation
- Scaling Issues: The emitters map is local to the application instance. If multiple instances of the application are running in a Kubernetes cluster, a request may be routed to an instance that does not hold a given client’s SseEmitter, causing message loss.
- Load Balancer Problem: Since Kubernetes LoadBalancer distributes requests across multiple instances, a client may get connected to one instance, but messages may be published from another, breaking the real-time connection.
- Memory & Connection Management: Keeping all emitters in memory can cause resource leaks if not managed properly.
Solution: Using Redis Pub/Sub for Consistency
To ensure real-time updates reach all instances of the application, I used Redis Pub/Sub. Instead of storing emitters in a local map, I publish messages to a Redis topic, and all instances listen for events on that Redis topic.
You can view this article for setting up Redis in your k8s cluster and configuring your Spring Boot application to connect to Redis.
I ran the below commands using helm to get Redis installed on Minikube.
helm repo add bitnami https://charts.bitnami.com/bitnami
helm repo update
# Install Redis with name 'my-redis' in redis namespace
helm install my-redis bitnami/redis --namespace redis
Create SSE Service
@Service
@Log4j2
public class SSEService implements MessageListener {
private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();
@Autowired
private RedisTemplate<String, String> redisTemplate;
// Called when a new SSE connection is established
public SseEmitter handler(String clientID) {
SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);
emitters.put(clientID, emitter);
emitter.onCompletion(() -> emitters.remove(clientID));
emitter.onTimeout(() -> emitters.remove(clientID));
return emitter;
}
// Publish an event by sending it to Redis
public void publishToEmitter(String clientID, String message) {
// Publish the message in a JSON or delimited format (including the clientID)
String payload = clientID + ":" + message;
redisTemplate.convertAndSend("sse-channel", payload);
}
// This method is invoked when a message is received from Redis
@Override
public void onMessage(Message message, byte[] pattern) {
String payload = new String(message.getBody(), StandardCharsets.UTF_8);
// Assuming payload is in format "clientID:message"
int index = payload.indexOf(":");
if (index > 0) {
String clientID = payload.substring(0, index);
String eventMessage = payload.substring(index + 1);
SseEmitter emitter = emitters.get(clientID);
if (emitter != null) {
try {
emitter.send(SseEmitter.event().data(eventMessage));
} catch (IOException e) {
emitter.completeWithError(e);
emitters.remove(clientID);
}
}
}
}
// Optionally, provide a method to retrieve all emitters if needed
public Collection<SseEmitter> getAllEmitters() {
return emitters.values();
}
}
Updated Controller
@RestController
@Log4j2
public class SSEController {
@Autowired
private SSEService sseService;
@CrossOrigin(origins = "*")
@GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter streamEvents(@RequestParam(name = "clientID") String clientID) {
return sseService.handler(clientID);
}
@CrossOrigin(origins = "*")
@GetMapping(value = "/publish")
public String publish(@RequestParam(name = "clientID") String clientID) {
long timestamp = System.currentTimeMillis();
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String formattedTime = sdf.format(new Date(timestamp));
sseService.publishToEmitter(clientID, "Hello at timestamp " + formattedTime);
return "Published to emitter: " + clientID + " at: " + formattedTime;
}
}
Scalability
Using Redis Pub/Sub makes this system horizontally scalable. Additional application replicas can be deployed on Kubernetes, and they will all stay in sync because Redis acts as a centralized event distributor. This ensures:
- Consistent client message delivery regardless of which instance they are connected to.
- Automatic load balancing without requiring sticky sessions.
- High availability because the system is not dependent on a single application instance.
Deploying on Kubernetes with Minikube
I have used helm for deploying my services. You can checkout the helm chart in this github repo
And you can checkout the full Spring Boot source code in this github repo
Conclusion
By leveraging Redis Pub/Sub, I ensure that SSE messages reach all instances of our Spring Boot application running on Kubernetes. This setup ensures high availability, scalability, and resilience in a microservices environment. Deploying on Minikube with a LoadBalancer further helps in testing real-world scenarios before moving to a production environment.
Let me know if you found this helpful or have any questions!
