1. Introduction
In this article, we’ll see a common mistake that developers make while using Spring Webflux. Spring Webflux is a non-blocking web framework built from the ground up to take advantage of multi-core, next-generation processors and handle massive concurrent connections.
Since it’s a non-blocking framework the threads shouldn’t be blocked. Let’s explore this in more detail.
2. Spring Webflux Threading Model
To understand this issue better we need to understand the threading model of Spring Webflux.
In Spring Webflux, a small pool of worker threads handles incoming requests. This contrasts with the Servlet model where each request gets a dedicated thread. Hence, the framework is protective of what happens on these request-accepting threads.
With that understanding in mind, let’s dive deep into the main focus of this article.
3. Understanding IllegalStateException With Thread Blocking
Let’s understand with the help of an example when and why we get the error “java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread” in Spring Webflux.
Let’s take an example of a file-searching API. This API reads a file from the file system and searches for the text provided by the user within the file.
3.1. File Service
Let’s first define a FileService class which reads a file’s content as a string:
@Service
public class FileService {
@Value("${files.base.dir:/tmp/bael-7724}")
private String filesBaseDir;
public Mono<String> getFileContentAsString(String fileName) {
return DataBufferUtils.read(Paths.get(filesBaseDir + "/" + fileName), DefaultDataBufferFactory.sharedInstance, DefaultDataBufferFactory.DEFAULT_INITIAL_CAPACITY)
.map(dataBuffer -> dataBuffer.toString(StandardCharsets.UTF_8))
.reduceWith(StringBuilder::new, StringBuilder::append)
.map(StringBuilder::toString);
}
}
It’s worth noting that FileService reads the file from the file system reactively (asynchronously).
3.2. File Content Search Service
We’re ready to leverage this FileService to write a file-searching service:
@Service
public class FileContentSearchService {
@Autowired
private FileService fileService;
public Mono<Boolean> blockingSearch(String fileName, String searchTerm) {
String fileContent = fileService
.getFileContentAsString(fileName)
.doOnNext(content -> ThreadLogger.log("1. BlockingSearch"))
.block();
boolean isSearchTermPresent = fileContent.contains(searchTerm);
return Mono.just(isSearchTermPresent);
}
}
The file-searching service returns a boolean depending on whether the search term is found in the file. For this, we call the getFileContentAsString() method of FileService. Since we get the result asynchronously, i.e., as a Mono
3.3. Files Controller
Finally, we’ve got the FileController which makes use of the FileContentSearchService‘s blockingSearch() method:
@RestController
@RequestMapping("bael7724/v1/files")
public class FileController {
...
@GetMapping(value = "/{name}/blocking-search")
Mono<Boolean> blockingSearch(@PathVariable("name") String fileName, @RequestParam String term) {
return fileContentSearchService.blockingSearch(fileName, term);
}
}
3.4. Reproducing the Exception
We can observe that the Controller calls the method of FileContentSearchService which in turn calls the block() method. Since this was on a request-accepting thread, if we call our API in the current arrangement, we would encounter the notorious exception we’re after:
12:28:51.610 [reactor-http-epoll-2] ERROR o.s.b.a.w.r.e.AbstractErrorWebExceptionHandler - [ea98e542-1] 500 Server Error for HTTP GET "/bael7724/v1/files/a/blocking-search?term=a"
java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-epoll-2
at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:86)
Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException:
Error has been observed at the following site(s):
*__checkpoint ⇢ com.baeldung.filters.TraceWebFilter [DefaultWebFilterChain]
*__checkpoint ⇢ com.baeldung.filters.ExceptionalTraceFilter [DefaultWebFilterChain]
*__checkpoint ⇢ HTTP GET "/bael7724/v1/files/a/blocking-search?term=a" [ExceptionHandlingWebHandler]
Original Stack Trace:
at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:86)
at reactor.core.publisher.Mono.block(Mono.java:1712)
at com.baeldung.bael7724.service.FileContentSearchService.blockingSearch(FileContentSearchService.java:20)
at com.baeldung.bael7724.controller.FileController.blockingSearch(FileController.java:35)
at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103)
at java.base/java.lang.reflect.Method.invoke(Method.java:580)
3.5. The Root Cause
The root cause of this exception is calling block() on the request-accepting thread. In our sample code above, the block() method is being called on one of the threads from the thread pool which accepts the request. Specifically, on the threads marked as “non-blocking operations only”, i.e., threads implementing Reactor’s NonBlocking marker interface, like those started by Schedulers.parallel().
4. Solution
Let’s now look at what can be done to address this exception.
4.1. Embracing the Reactive Operations
The idiomatic approach is to use reactive operations instead of calling block(). Let’s update the code to make use of the map() operation to translate the String into Boolean:
public Mono<Boolean> nonBlockingSearch(String fileName, String searchTerm) {
return fileService.getFileContentAsString(fileName)
.doOnNext(content -> ThreadLogger.log("1. NonBlockingSearch"))
.map(content -> content.contains(searchTerm))
.doOnNext(content -> ThreadLogger.log("2. NonBlockingSearch"));
}
We’ve thus eliminated the need to call block() altogether. When we run the above method we notice the following thread context:
[1. NonBlockingSearch] ThreadName: Thread-4, Time: 2024-06-17T07:40:59.506215299Z
[2. NonBlockingSearch] ThreadName: Thread-4, Time: 2024-06-17T07:40:59.506361786Z
[1. In Controller] ThreadName: Thread-4, Time: 2024-06-17T07:40:59.506465805Z
[2. In Controller] ThreadName: Thread-4, Time: 2024-06-17T07:40:59.506543145Z
The above log statements indicate that we’ve performed the operations on the same thread pool which is accepting the request.
Notably, even though we didn’t encounter the exception it’s better to run the I/O operations such as reading from a file on a different thread pool.
4.2. Blocking on the Bounded Elastic Thread Pool
Let’s say we can’t avoid block() for some reason. Then how do we go about this? We concluded that the exception occurred as we called block() on the request-accepting thread pool. Hence, to call block() we need to switch the thread pool. Let’s see how we can do this:
public Mono<Boolean> workableBlockingSearch(String fileName, String searchTerm) {
return Mono.just("")
.doOnNext(s -> ThreadLogger.log("1. WorkableBlockingSearch"))
.publishOn(Schedulers.boundedElastic())
.doOnNext(s -> ThreadLogger.log("2. WorkableBlockingSearch"))
.map(s -> fileService.getFileContentAsString(fileName)
.block()
.contains(searchTerm))
.doOnNext(s -> ThreadLogger.log("3. WorkableBlockingSearch"));
}
To switch the thread pool, Spring Webflux provides two operations publishOn() and subscribeOn(). We’ve used publishOn() which changes the thread for the operations that come after publishOn(), without affecting the subscription or the upstream operations. Since the thread pool is now switched to the bounded elastic, we can call block().
Now, if we run the workableBlockingSearch() method we’ll get the following thread context:
[1. WorkableBlockingSearch] ThreadName: parallel-2, Time: 2024-06-17T07:40:59.440562518Z
[2. WorkableBlockingSearch] ThreadName: boundedElastic-1, Time: 2024-06-17T07:40:59.442161018Z
[3. WorkableBlockingSearch] ThreadName: boundedElastic-1, Time: 2024-06-17T07:40:59.442891230Z
[1. In Controller] ThreadName: boundedElastic-1, Time: 2024-06-17T07:40:59.443058091Z
[2. In Controller] ThreadName: boundedElastic-1, Time: 2024-06-17T07:40:59.443181770Z
We can see that from number 2 onwards the operations have indeed happened on the bounded elastic thread pool hence, we didn’t get IllegalStateException.
4.3. Caveats
Let’s look at some of the caveats to this solution with the block.
There are many ways we can go wrong with calling block(). Let’s take one example where even though we use a Scheduler to switch the thread context it doesn’t behave the way we expect it to:
public Mono<Boolean> incorrectUseOfSchedulersSearch(String fileName, String searchTerm) {
String fileContent = fileService.getFileContentAsString(fileName)
.doOnNext(content -> ThreadLogger.log("1. IncorrectUseOfSchedulersSearch"))
.publishOn(Schedulers.boundedElastic())
.doOnNext(content -> ThreadLogger.log("2. IncorrectUseOfSchedulersSearch"))
.block();
boolean isSearchTermPresent = fileContent.contains(searchTerm);
return Mono.just(isSearchTermPresent);
}
In the above code sample, we’ve used publishOn() as recommended in the solution but the block() method is still causing the exception. When we run the above code, we’ll get the following logs:
[1. IncorrectUseOfSchedulersSearch] ThreadName: Thread-4, Time: 2024-06-17T08:57:02.490298417Z
[2. IncorrectUseOfSchedulersSearch] ThreadName: boundedElastic-1, Time: 2024-06-17T08:57:02.491870410Z
14:27:02.495 [parallel-1] ERROR o.s.b.a.w.r.e.AbstractErrorWebExceptionHandler - [53e4bce1] 500 Server Error for HTTP GET "/bael7724/v1/files/robots.txt/incorrect-use-of-schedulers-search?term=r-"
java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread parallel-1
at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:86)
Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException:
Error has been observed at the following site(s):
*__checkpoint ⇢ com.baeldung.filters.TraceWebFilter [DefaultWebFilterChain]
*__checkpoint ⇢ com.baeldung.filters.ExceptionalTraceFilter [DefaultWebFilterChain]
*__checkpoint ⇢ HTTP GET "/bael7724/v1/files/robots.txt/incorrect-use-of-schedulers-search?term=r-" [ExceptionHandlingWebHandler]
Original Stack Trace:
at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:86)
at reactor.core.publisher.Mono.block(Mono.java:1712)
at com.baeldung.bael7724.service.FileContentSearchService.incorrectUseOfSchedulersSearch(FileContentSearchService.java:64)
at com.baeldung.bael7724.controller.FileController.incorrectUseOfSchedulersSearch(FileController.java:48)
at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103)
at java.base/java.lang.reflect.Method.invoke(Method.java:580)
This indicates that the second log statement did indeed run on the bounded elastic thread pool. However, we still encountered the exception. The reason is the block() is still running on the same request accepting thread pool.
Let’s look at another caveat. Even though we’re switching the thread pool, we cannot use the parallel thread pool, i.e., Schedulers.parallel(). As mentioned earlier certain thread pools don’t allow invoking block() on their threads – parallel thread pool is one of them.
Finally, we’ve only used Schedulers.boundedElastic() in our examples. Instead, we could have also used any custom thread pool via Schedulers.fromExecutorService().
5. Conclusion
In conclusion, to effectively address the issue of IllegalStateException in Spring Webflux when using blocking operations like block(), we should adopt a non-blocking, reactive approach. By leveraging reactive operators such as map(), we can perform operations on the same reactive thread pool, eliminating the need for explicit block(). If block() cannot be avoided, switching the execution context to a boundedElastic scheduler or a custom thread pool using publishOn() can isolate these operations from the reactive request-accepting thread pool, thus preventing exceptions.
It’s essential to be aware of thread pools that don’t support blocking calls and to ensure that the correct context switching is applied to maintain the application’s stability and performance.
As always, the source code used in this article is available over on GitHub.