Embracing Reactive Programming Using Queues
Image by Pixabay“The goal as a company is to have customer service that is not just the best, but legendary.” — Sam Walton
Introduction
Data pipelines often require scaling between peak loads and downtime. Automated systems can reduce resource utilization by 95% compared to older, legacy systems such which often require manual intervention to scale the underlying workers and performance tuning for bolt parallelism and tasks. This can be tedious especially when the data throughput is not stable and can vary significantly based on the time, day, and month.
Without the capability to auto scale based on throughput, the pipelines needed to be scaled to process the max throughput — increasing the computing cost unnecessarily when the throughput is less than the max. Here’s a sample topology, with each bolt represented as a circle. The size of the circle indicates the parallelism configuration to handle the throughput based on the bolt latency. The color indicates where the pipeline could be choking.
Shift to Reactive Programming
Reactive programming is a novel approach in software development. But what exactly is it? It is a declarative, event-driven programming paradigm concerned with data streams and the propagation of change. In reactive programming, the availability of new information drives the logic forward rather than having the thread-of-execution control the flow.
It supports breaking down a request into various steps which can be executed asynchronously. These steps can be wired together to form larger workflows in a non-blocking manner. The fundamental idea behind reactive programming is that events are data and data are events.
Reactive programming offers several advantages:
- Efficient Resource Utilization: Reactive programming is designed for maximum efficiency by only reacting when necessary, saving computational resources, whereas a continuous query model can lead to unnecessary computations and inefficient resource utilization.
- Backpressure Support: Reactive programming inherently supports backpressure, which allows consumers to control how much data they receive. This prevents system overload and enhances overall performance.
- Scalability: Reactive programming is inherently scalable. It allows for more efficient use of computing resources by only reacting to inputs or changes, rather than continuously polling for changes. This makes it easier to scale up or down based on the system load.
- Declarative Programming Model: Reactive programming uses a declarative programming model. This makes the code easier to read, write, and debug.
Migration Strategy
Message Processing Stream — Reactive stream framework that can consume and process from message queues. It can process messages in batches when there is high volume and provide backpressure capability when the API is not able to scale or when the underlying resource is choking.
Setup APIs — Invokes HTTP request and process response in a non-blocking fashion. The application jar can be run anywhere, enabling the application to leverage cloud elasticity based on CPU utilization of the application and thus saving on cost.
This enables us to separate the message queue consumer and the execution of the task pipeline. This is key to scale the two independently and support retry, timeouts and error processing functionalities.
Here’s a sample snippet on how such a system can be implemented, written in java.
// Create the consumer with custom properties
KafkaReceiver<String, String> receiver = KafkaReceiver.create(getMessageQueueOptions()));
// Decouple Consumer pipeline with event handling business logic
receiver.receive()
.publishOn(Schedulers.boundedElastic())
.doOnError(error -> log.error(String.format("Failed to consume events from topic%s", error.getClass().getName())))
.map(this::createEvent)
.flatMap(this::handleEvent, 100);
// Process the event by invoking API for execution and commit the event, irrespective of success or failure return the original event
protected Mono<MessageEnvelope> handleEvent(MessageEnvelope event) {
return Mono.just(event)
.<MessageEnvelope>handle((MessageEnvelope, sink) -> {
if (!MessageEnvelope.isValidEvent()) {
log.error(String.format("Incoming payload is invalid dropping event %s", MessageEnvelope.getMessageRecord().key()));
} else {
log.info(String.format("ValidatedEventprocessingrequest%s"));
sink.next(MessageEnvelope);
}
})
.flatMap(MessageEnvelope -> {
log.info(String.format("Sending setuprequest for event request_id=%s", MessageEnvelope.getRequestId()));
return APIService.setup(MessageEnvelope, getHeaderMap(MessageEnvelope.getMessageRecord().headers()));
})
.doOnNext(processedEvent -> log.info(String.format("Successfullyprocessedevent:%s", processedEvent.getStatus())))
.doOnError(ex -> log.error(String.format("Errorprocessingeventat%s", ex.getMessage())))
.onErrorResume(ex -> {
log.error(String.format("Observed Exception inhandleEvent%s", ex.getMessage()));
return Mono.empty();
})
.then(Mono.just(event));
}
// Create a webclient with custom configuration
HttpClient httpClient = HttpClient.create()
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000);
ClientHttpConnector connector = new ReactorClientHttpConnector(httpClient);
WebClient webClient = WebClient.builder()
.clientConnector(connector)
.defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
.build();
public Mono<String> setup(String requestPayload, Map<String, String> headers) {
return webClient.post()
.uri(getCompleteURL())
.httpRequest(httpRequest -> {
HttpClientRequest reactorRequest = httpRequest.getNativeRequest();
reactorRequest.responseTimeout(Duration.ofMillis(3000));
})
.headers(getHttpHeaders(headers, context))
.bodyValue(requestPayload)
.exchangeToMono(clientResponse ->
clientResponse.bodyToMono(String.class)
.doOnNext(response -> log.info(String.format("Setup API request was success %s", clientResponse.statusCode().value())))
)
.onErrorResume(ex -> {
log.info(String.format("Setup API request Failed with error %s", ex.getMessage()));
return Mono.just(requestPayload);
});
}
// Create a controller to accept request and call the stream pipeline underneath.
@PostMapping("/setup")
public Mono<ResponseEntity<String>> setup(@RequestBody String request, @RequestHeader Map<String, Object> headers) {
return Mono.just(request)
.map(request -> {
log.info(String.format("Received request for setup"));
return new SetupRequest(request);
})
.flatMap(requestPayload -> setupTaskPipeline.setup(requestPayload)
.map(response -> getHttpResponseEntity(response))
)
.onErrorReturn(ResponseEntity
.status(500)
.contentType(MediaType.APPLICATION_JSON)
.body("Request failed with unknown error")
);
}
// Build a reactor stream pipeline which invokes services based on service dependency and invoke each services in a non-blocking way
public Mono<SetupResponse> setup(SetupRequest request) {
return validationService.validate(request)
.flatMap(validatedRequest -> idGenerationService.invokeAPI(validatedRequest))
.flatMap(responseFromIDService -> classifierService.invokeAPI(responseFromIDService))
.flatMap(responseFromClassification-> persistanceService.invokeAPI(responseFromClassification))
.onErrorResume(ex -> Mono.just(updateErrorMessage(request)))
.flatMap(resp -> triggerErrorFlow(resp));
}Lessons Learned
Using reactive programming, we were able to increase the max throughput of one particular pipeline from 800 TPS to 2000 TPS, which can be further scaled by adding additional resources. This reduced the overall resource utilization by 95% with significant cost savings.
Conclusion
Utilizing reactive programming enabled us to effectively manage blocking IO operations (such as HTTP API), backpressure, thread handling/scheduling, and error handling. It necessitated a change in both our paradigm and mindset regarding how we design and manage the application’s flow. It was a learning curve for our team to grasp the necessary techniques and gradually integrate them into our extensive and continually evolving backend infrastructure. For systems that require automatic and extensive scaling, reactive programming has proven to be highly resource and cost effective.
Embracing Reactive Programming Using Kafka & Webflux was originally published in Walmart Global Tech Blog on Medium, where people are continuing the conversation by highlighting and responding to this story.
Introduction to Malware Binary Triage (IMBT) Course
Looking to level up your skills? Get 10% off using coupon code: MWNEWS10 for any flavor.
Enroll Now and Save 10%: Coupon Code MWNEWS10
Note: Affiliate link – your enrollment helps support this platform at no extra cost to you.
Article Link: Embracing Reactive Programming Using Queues | by Vikas Gowda | Walmart Global Tech Blog | Feb, 2025 | Medium