GeeCON 2019: Don’t be Homer Simpson with your Reactor!

Project Reactor is a very powerful tool in the developers' hands. But, the more powerful tools you use, the stronger safety rules are. Do you know all of them?

In this talk, we will learn about advanced Project Reactor debugging & testing techniques, which are helpful for anyone who works with Project Reactor. Because like the cockroaches can survive even the radiation, bugs can escape the most well-thought-out code.

Never heard about BlockHound?
Or you know about Backpressure, but not sure if it is handled correctly in your​ code?
Maybe you want to prepare you reactive code upfront to help your future self when something goes wrong in prod?
Or perhaps just watch how Reactor team members work with Reactor code and learn some neat tricks?
Then this talk is for you!

Transcript

1. Don’t be Homer Simpson 
 with your Reactor! Sergei Egorov Kraków, 15-17 May 2019 @bsideup
2. About me • Staff Engineer at Pivotal’s Spring R&D, working on Project Reactor ⚛ • Berlin Spring User Group co-organizer • Testcontainers co-maintainer • Developer tools geek • … although spent most of my career building
 highly scalable backend services @bsideup
3. My first experience with
 reactive programming was like…
4. @bsideup
5. @bsideup
6. Mono.fromRunnable()
7. Mono.fromRunnable() @bsideup
8. Mono.fromRunnable() My code @bsideup
9. Mono.fromRunnable() My code My code @bsideup
10. Mono.fromRunnable() My code My code My code Work! @bsideup
11. Mono.fromRunnable().subscribe() @bsideup
12. 1 week later…
13. I agree with Heinrich Apfelmus that the essence of functional reactive programming is to specify the dynamic behavior of a value completely at the time of declaration. “Reactive Sir” @bsideup
14. 1 week of production later…
15. @bsideup
16. What could go wrong
17. Whoa #1: exceptions
18. // com/example/demo/Example.java Flux.range(0, 5) .single() .subscribeOn(Schedulers.parallel()) .subscribe(); @bsideup
19. // com/example/demo/Example.java Flux.range(0, 5) .single() .subscribeOn(Schedulers.parallel()) .subscribe(); java.lang.IndexOutOfBoundsException: Source emitted more than one item at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(MonoSingle.java:129) at reactor.core.publisher.FluxRange$RangeSubscription.fastPath(FluxRange.java:129) at reactor.core.publisher.FluxRange$RangeSubscription.request(FluxRange.java:107) at reactor.core.publisher.MonoSingle$SingleSubscriber.request(MonoSingle.java:94) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.trySchedule(MonoSubscribeOn.java:186) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.onSubscribe(MonoSubscribeOn.java:131) at reactor.core.publisher.MonoSingle$SingleSubscriber.onSubscribe(MonoSingle.java:114) at reactor.core.publisher.FluxRange.subscribe(FluxRange.java:68) at reactor.core.publisher.MonoSingle.subscribe(MonoSingle.java:58) at reactor.core.publisher.Mono.subscribe(Mono.java:3711) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.run(MonoSubscribeOn.java:123) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) @bsideup
20. // com/example/demo/Example.java Flux.range(0, 5) .single() .subscribeOn(Schedulers.parallel()) .subscribe(); java.lang.IndexOutOfBoundsException: Source emitted more than one item at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(MonoSingle.java:129) at reactor.core.publisher.FluxRange$RangeSubscription.fastPath(FluxRange.java:129) at reactor.core.publisher.FluxRange$RangeSubscription.request(FluxRange.java:107) at reactor.core.publisher.MonoSingle$SingleSubscriber.request(MonoSingle.java:94) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.trySchedule(MonoSubscribeOn.java:186) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.onSubscribe(MonoSubscribeOn.java:131) at reactor.core.publisher.MonoSingle$SingleSubscriber.onSubscribe(MonoSingle.java:114) at reactor.core.publisher.FluxRange.subscribe(FluxRange.java:68) at reactor.core.publisher.MonoSingle.subscribe(MonoSingle.java:58) at reactor.core.publisher.Mono.subscribe(Mono.java:3711) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.run(MonoSubscribeOn.java:123) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) @bsideup
21. // com/example/demo/Example.java Flux.range(0, 5) .single() .subscribeOn(Schedulers.parallel()) .subscribe(); java.lang.IndexOutOfBoundsException: Source emitted more than one item at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(MonoSingle.java:129) at reactor.core.publisher.FluxRange$RangeSubscription.fastPath(FluxRange.java:129) at reactor.core.publisher.FluxRange$RangeSubscription.request(FluxRange.java:107) at reactor.core.publisher.MonoSingle$SingleSubscriber.request(MonoSingle.java:94) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.trySchedule(MonoSubscribeOn.java:186) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.onSubscribe(MonoSubscribeOn.java:131) at reactor.core.publisher.MonoSingle$SingleSubscriber.onSubscribe(MonoSingle.java:114) at reactor.core.publisher.FluxRange.subscribe(FluxRange.java:68) at reactor.core.publisher.MonoSingle.subscribe(MonoSingle.java:58) at reactor.core.publisher.Mono.subscribe(Mono.java:3711) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.run(MonoSubscribeOn.java:123) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) Which source? @bsideup
22. // com/example/demo/Example.java Flux.range(0, 5) .single() .subscribeOn(Schedulers.parallel()) .subscribe(); java.lang.IndexOutOfBoundsException: Source emitted more than one item at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(MonoSingle.java:129) at reactor.core.publisher.FluxRange$RangeSubscription.fastPath(FluxRange.java:129) at reactor.core.publisher.FluxRange$RangeSubscription.request(FluxRange.java:107) at reactor.core.publisher.MonoSingle$SingleSubscriber.request(MonoSingle.java:94) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.trySchedule(MonoSubscribeOn.java:186) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.onSubscribe(MonoSubscribeOn.java:131) at reactor.core.publisher.MonoSingle$SingleSubscriber.onSubscribe(MonoSingle.java:114) at reactor.core.publisher.FluxRange.subscribe(FluxRange.java:68) at reactor.core.publisher.MonoSingle.subscribe(MonoSingle.java:58) at reactor.core.publisher.Mono.subscribe(Mono.java:3711) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.run(MonoSubscribeOn.java:123) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) Which source? Where is “com.example.demo” package? @bsideup
23. // com/example/demo/Example.java Flux.range(0, 5) .single() .subscribeOn(Schedulers.parallel()) .subscribe(); java.lang.IndexOutOfBoundsException: Source emitted more than one item at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(MonoSingle.java:129) at reactor.core.publisher.FluxRange$RangeSubscription.fastPath(FluxRange.java:129) at reactor.core.publisher.FluxRange$RangeSubscription.request(FluxRange.java:107) at reactor.core.publisher.MonoSingle$SingleSubscriber.request(MonoSingle.java:94) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.trySchedule(MonoSubscribeOn.java:186) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.onSubscribe(MonoSubscribeOn.java:131) at reactor.core.publisher.MonoSingle$SingleSubscriber.onSubscribe(MonoSingle.java:114) at reactor.core.publisher.FluxRange.subscribe(FluxRange.java:68) at reactor.core.publisher.MonoSingle.subscribe(MonoSingle.java:58) at reactor.core.publisher.Mono.subscribe(Mono.java:3711) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.run(MonoSubscribeOn.java:123) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) Which source? Where is “com.example.demo” package? Why they closed Google Inbox? @bsideup
24. Assembly vs Subscription
25. Flux.range(0, 5) Assembly @bsideup
26. Flux.range(0, 5) .filter(it -> it > 3) Assembly @bsideup
27. Flux.range(0, 5) .filter(it -> it > 3) .map(it -> it.toString()) Assembly @bsideup
28. Flux.range(0, 5) .filter(it -> it > 3) .map(it -> it.toString()) .single() Assembly @bsideup
29. Flux.range(0, 5) .filter(it -> it > 3) .map(it -> it.toString()) .single() .subscribeOn(Schedulers.parallel()) Assembly @bsideup
30. Flux.range(0, 5) .filter(it -> it > 3) .map(it -> it.toString()) .single() .subscribeOn(Schedulers.parallel()) .subscribe() Assembly @bsideup
31. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 @bsideup
32. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 * = subscriber @bsideup
33. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 @bsideup
34. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 @bsideup
35. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 @bsideup
36. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 @bsideup
37. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 .onSubscribe( ) * = subscription @bsideup
38. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 @bsideup
39. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 @bsideup
40. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 @bsideup
41. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 @bsideup
42. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 .request(max) @bsideup
43. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 .request(max) @bsideup
44. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 .request(max) @bsideup
45. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 .request(max) @bsideup
46. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 .request(max) @bsideup
47. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 .request(max) @bsideup
48. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Execution current thread parallel-1 .onNext(0) @bsideup
49. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Execution current thread parallel-1 .onNext(0) @bsideup
50. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Execution current thread parallel-1 .onNext(0) @bsideup
51. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Execution current thread parallel-1 .onNext(0) @bsideup
52. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Execution current thread parallel-1 .onNext(0) @bsideup
53. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Execution current thread parallel-1 .onNext(1) @bsideup
54. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Execution current thread parallel-1 .onNext(1) @bsideup
55. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Execution current thread parallel-1 .onNext(1) java.lang.IndexOutOfBoundsException: Source emitted more than one item at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(Mono at reactor.core.publisher.FluxRange$RangeSubscription.fastPath(Flux at reactor.core.publisher.FluxRange$RangeSubscription.request(FluxR at reactor.core.publisher.MonoSingle$SingleSubscriber.request(Mono at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber. at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber. at reactor.core.publisher.MonoSingle$SingleSubscriber.onSubscribe(M at reactor.core.publisher.FluxRange.subscribe(FluxRange.java:68) at reactor.core.publisher.MonoSingle.subscribe(MonoSingle.java:58) at reactor.core.publisher.Mono.subscribe(Mono.java:3711) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber. @bsideup
56. Demo http://eskipaper.com/homer-jay-simpson-cartoon.html
57. How ReactorDebugAgent works? @bsideup
58. // com/example/demo/Example.java Flux.range(0, 5) .single() .subscribeOn(Schedulers.parallel()) .subscribe(); @bsideup
59. // com/example/demo/Example.java Flux.range(0, 5) .single() .subscribeOn(Schedulers.parallel()) .subscribe(); @bsideup
60. // com/example/demo/Example.java Flux.range(0, 5) .checkpoint("Example.java:16") .single() .checkpoint("Example.java:17") .subscribeOn(Schedulers.parallel()) .checkpoint(“Example.java:18") .subscribe(); @bsideup
61. Demo outcomes • Use .checkpoint(“something”) to “mark” reactive “milestones” • Read about Hooks.onOperatorDebug()… • … but use reactor-tools’ ReactorDebugAgent (works in prod too) • https://spring.io/blog/2019/03/06/flight-of-the-flux-1-assembly-vs- subscription - great article from Simon Basle about the internals @bsideup
62. Whoa #2: Blocking calls! https://blog.tfd.co.uk/2010/10/15/jackrabbit-performance/ @bsideup
63. How Reactor schedules tasks @bsideup
64. Default thread pools Schedulers.parallel() - N threads, where N matches the CPUs count. Schedulers.single() - 1 thread handling all submitted tasks Schedulers.elastic() - dynamic, thread caching pool @bsideup
65. Mono.delay(ofSeconds(1)) @bsideup
66. Mono.delay(ofSeconds(1)) Mono.delay(ofSeconds(1), Schedulers.parallel()) @bsideup
67. Mono.delay(ofSeconds(1)) Mono.delay(ofSeconds(1), Schedulers.parallel()) Every non-instant operation 
 runs on the parallel scheduler
 by default! @bsideup
68. parallel-1 parallel-2 parallel-3 parallel-4 Non-blocking execution @bsideup
69. parallel-1 parallel-2 parallel-3 parallel-4 Non-blocking execution @bsideup
70. parallel-1 parallel-2 parallel-3 parallel-4 Non-blocking execution @bsideup
71. parallel-1 parallel-2 parallel-3 parallel-4 Non-blocking execution @bsideup
72. parallel-1 parallel-2 parallel-3 parallel-4 Non-blocking execution @bsideup
73. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution @bsideup
74. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution @bsideup
75. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution @bsideup
76. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution @bsideup
77. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution @bsideup
78. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution @bsideup
79. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution @bsideup
80. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution @bsideup
81. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution @bsideup
82. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution ⚠ no more tasks scheduled 
 until this task returns @bsideup
83. How to fix?
84. Use non-blocking APIs, or…
85. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
86. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
87. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
88. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
89. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
90. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
91. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
92. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
93. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
94. Demo http://eskipaper.com/homer-jay-simpson-cartoon.html
95. How BlockHound works? @bsideup
96. public class Thread implements Runnable { public static native void sleep(long millis); } @bsideup
97. public class Thread implements Runnable { private static native void $$BlockHound$$_sleep(long millis); public static void sleep(long millis) { $$BlockHound$$_sleep(millis); } } @bsideup
98. public class Thread implements Runnable { private static native void $$BlockHound$$_sleep(long millis); public static void sleep(long millis) { reactor.BlockHoundRuntime.checkBlocking( "java.lang.Thread", "sleep", /*method modifiers*/ ); $$BlockHound$$_sleep(millis); } } @bsideup
99. public class Thread implements Runnable { private static native void $$BlockHound$$_sleep(long millis); public static void sleep(long millis) { reactor.BlockHoundRuntime.checkBlocking( "java.lang.Thread", "sleep", /*method modifiers*/ ); $$BlockHound$$_sleep(millis); } } @bsideup
100. Blocking check @bsideup
101. Blocking check 1. Get or tag a current thread • JVMTI’s built-in mechanism to tag objects, calls nonBlockingThreadPredicate on creation and caches the result forever @bsideup
102. Blocking check 1. Get or tag a current thread • JVMTI’s built-in mechanism to tag objects, calls nonBlockingThreadPredicate on creation and caches the result forever 2. Is it non-blocking? (the predicate returns “true”) • Very fast, O(1) check @bsideup
103. Blocking check 1. Get or tag a current thread • JVMTI’s built-in mechanism to tag objects, calls nonBlockingThreadPredicate on creation and caches the result forever 2. Is it non-blocking? (the predicate returns “true”) • Very fast, O(1) check 3. Walk the stacktrace until a frame marked with (dis)allowBlockingCallsInside • Makes it possible to whitelist loggers and other non-harmful blocking calls @bsideup
104. Stack @bsideup at sun.misc.Unsafe.park(Unsafe.java) at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175) at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836) at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997) at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304) at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231) at com.example.demo.BlockingCodeTest.lambda$testBlockingCode$1(BlockingCodeTest.java:24) at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:100) at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.onNext(FluxSubscribeOn.java:151) at reactor.core.publisher.FluxRange$RangeSubscription.fastPath(FluxRange.java:129) at reactor.core.publisher.FluxRange$RangeSubscription.request(FluxRange.java:107) at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.requestUpstream(FluxSubscribeOn.java:131) at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.onSubscribe(FluxSubscribeOn.java:124) at reactor.core.publisher.FluxRange.subscribe(FluxRange.java:68) at reactor.core.publisher.Flux.subscribe(Flux.java:7800) at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.run(FluxSubscribeOn.java:194) at reactor.core.scheduler.ReactorBlockHoundIntegration$Wrapper.run(ReactorBlockHoundIntegration.java:56) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748)
105. Stack @bsideup at sun.misc.Unsafe.park(Unsafe.java) at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175) at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836) at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997) at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304) at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231) at com.example.demo.BlockingCodeTest.lambda$testBlockingCode$1(BlockingCodeTest.java:24) at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:100) at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.onNext(FluxSubscribeOn.java:151) at reactor.core.publisher.FluxRange$RangeSubscription.fastPath(FluxRange.java:129) at reactor.core.publisher.FluxRange$RangeSubscription.request(FluxRange.java:107) at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.requestUpstream(FluxSubscribeOn.java:131) at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.onSubscribe(FluxSubscribeOn.java:124) at reactor.core.publisher.FluxRange.subscribe(FluxRange.java:68) at reactor.core.publisher.Flux.subscribe(Flux.java:7800) at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.run(FluxSubscribeOn.java:194) at reactor.core.scheduler.ReactorBlockHoundIntegration$Wrapper.run(ReactorBlockHoundIntegration.java:56) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748)
106. Stack @bsideup at sun.misc.Unsafe.park(Unsafe.java) at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175) at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836) at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997) at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304) at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231) at com.example.demo.BlockingCodeTest.lambda$testBlockingCode$1(BlockingCodeTest.java:24) at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:100) at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.onNext(FluxSubscribeOn.java:151) at reactor.core.publisher.FluxRange$RangeSubscription.fastPath(FluxRange.java:129) at reactor.core.publisher.FluxRange$RangeSubscription.request(FluxRange.java:107) at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.requestUpstream(FluxSubscribeOn.java:131) at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.onSubscribe(FluxSubscribeOn.java:124) at reactor.core.publisher.FluxRange.subscribe(FluxRange.java:68) at reactor.core.publisher.Flux.subscribe(Flux.java:7800) at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.run(FluxSubscribeOn.java:194) at reactor.core.scheduler.ReactorBlockHoundIntegration$Wrapper.run(ReactorBlockHoundIntegration.java:56) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748)
107. Stack @bsideup at sun.misc.Unsafe.park(Unsafe.java) at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175) at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836) at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997) at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304) at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231) at com.example.demo.BlockingCodeTest.lambda$testBlockingCode$1(BlockingCodeTest.java:24) at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:100) at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.onNext(FluxSubscribeOn.java:151) at reactor.core.publisher.FluxRange$RangeSubscription.fastPath(FluxRange.java:129) at reactor.core.publisher.FluxRange$RangeSubscription.request(FluxRange.java:107) at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.requestUpstream(FluxSubscribeOn.java:131) at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.onSubscribe(FluxSubscribeOn.java:124) at reactor.core.publisher.FluxRange.subscribe(FluxRange.java:68) at reactor.core.publisher.Flux.subscribe(Flux.java:7800) at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.run(FluxSubscribeOn.java:194) at reactor.core.scheduler.ReactorBlockHoundIntegration$Wrapper.run(ReactorBlockHoundIntegration.java:56) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748)
108. Stack @bsideup at sun.misc.Unsafe.park(Unsafe.java) at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175) at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836) at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997) at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304) at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231) at com.example.demo.BlockingCodeTest.lambda$testBlockingCode$1(BlockingCodeTest.java:24) at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:100) at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.onNext(FluxSubscribeOn.java:151) at reactor.core.publisher.FluxRange$RangeSubscription.fastPath(FluxRange.java:129) at reactor.core.publisher.FluxRange$RangeSubscription.request(FluxRange.java:107) at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.requestUpstream(FluxSubscribeOn.java:131) at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.onSubscribe(FluxSubscribeOn.java:124) at reactor.core.publisher.FluxRange.subscribe(FluxRange.java:68) at reactor.core.publisher.Flux.subscribe(Flux.java:7800) at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.run(FluxSubscribeOn.java:194) at reactor.core.scheduler.ReactorBlockHoundIntegration$Wrapper.run(ReactorBlockHoundIntegration.java:56) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) Schedulers.onScheduleHook("BlockHound", Wrapper::new); builder.disallowBlockingCallsInside(Wrapper.class.getName(), "run");
109. Demo outcomes • Blocking calls are bad, m’kay? • They may sneak into your production system • Use https://github.com/reactor/BlockHound to detect them • Supports multiple frameworks (Reactor, RxJava, etc) • … and maybe even Kotlin: 
 https://github.com/Kotlin/kotlinx.coroutines/issues/1031 • Use a dedicated pool for the necessary blocking calls, or schedule them on the Schedulers.elastic() built-in pool if they happen rarely @bsideup
110. “I use async APIs and I am safe!”
111. Yeah… sure.
112. Or…
113. Are you sure?
114. KafkaProducer#send(ProducerRecord,Callback) @bsideup
115. KafkaProducer#send(ProducerRecord,Callback) “Asynchronously send a record to a topic and invoke the provided callback when the send has been acknowledged.” - Javadoc @bsideup
116. KafkaProducer#send(ProducerRecord,Callback) “Asynchronously send a record to a topic and invoke the provided callback when the send has been acknowledged.” - Javadoc java.lang.Error: Blocking call! java.lang.Object#wait at reactor.BlockHound$Builder.lambda$new$0(BlockHound.java:154) at reactor.BlockHound$Builder.lambda$install$8(BlockHound.java:254) at reactor.BlockHoundRuntime.checkBlocking(BlockHoundRuntime.java:43) at java.lang.Object.wait(Object.java) at org.apache.kafka.clients.Metadata.awaitUpdate(Metadata.java:181) at org.apache.kafka.clients.producer.KafkaProducer.waitOnMetadata(KafkaProducer.java:938) at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:823) at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:803) @bsideup
117. KafkaProducer#send(ProducerRecord,Callback) “Asynchronously send a record to a topic and invoke the provided callback when the send has been acknowledged.” - Javadoc java.lang.Error: Blocking call! java.lang.Object#wait at reactor.BlockHound$Builder.lambda$new$0(BlockHound.java:154) at reactor.BlockHound$Builder.lambda$install$8(BlockHound.java:254) at reactor.BlockHoundRuntime.checkBlocking(BlockHoundRuntime.java:43) at java.lang.Object.wait(Object.java) at org.apache.kafka.clients.Metadata.awaitUpdate(Metadata.java:181) at org.apache.kafka.clients.producer.KafkaProducer.waitOnMetadata(KafkaProducer.java:938) at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:823) at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:803) https://issues.apache.org/jira/browse/KAFKA-3539 @bsideup
118. long remainingWaitMs = maxWaitMs; long elapsed; // Issue metadata requests until we have metadata for the topic or maxWaitTimeMs is exceeded. // In case we already have cached metadata for the topic, but the requested partition is greater // than expected, issue an update request only once. This is necessary in case the metadata // is stale and the number of partitions for this topic has increased in the meantime. do { log.trace("Requesting metadata update for topic {}.", topic); metadata.add(topic); int version = metadata.requestUpdate(); sender.wakeup(); try { metadata.awaitUpdate(version, remainingWaitMs); } catch (TimeoutException ex) { // Rethrow with original maxWaitMs to prevent logging exception with remainingWaitMs throw new TimeoutException("Failed to update metadata after " + maxWaitMs + " ms."); } cluster = metadata.fetch(); elapsed = time.milliseconds() - begin; if (elapsed >= maxWaitMs) throw new TimeoutException("Failed to update metadata after " + maxWaitMs + " ms."); if (cluster.unauthorizedTopics().contains(topic)) throw new TopicAuthorizationException(topic); remainingWaitMs = maxWaitMs - elapsed; partitionsCount = cluster.partitionCountForTopic(topic); } while (partitionsCount == null); waitOnMetadata(record.topic(), record.partition(), maxBlockTimeMs); Default is “60 seconds” @bsideup
119. @bsideup
120. How to instrument Reactor
121. Schedulers.onScheduleHook(fn) @bsideup
122. Schedulers.onScheduleHook(fn) @bsideup Schedulers.onScheduleHook("myHook", runnable -> { System.out.println("Before every scheduled runnable"); return () -> { System.out.println("Before execution"); runnable.run(); System.out.println("After execution"); }; });
123. Hooks.onEachOperator(fn) @bsideup Flux.range(0, 5) .map(it -> it.toString()) .single() .subscribeOn(Schedulers.parallel()) .subscribe();
124. Hooks.onEachOperator(fn) @bsideup Flux.range(0, 5) .map(it -> it.toString()) .single() .subscribeOn(Schedulers.parallel()) .subscribe();
125. Hooks.onEachOperator(fn) @bsideup Flux.range(0, 5) .map(it -> it.toString()) .single() .subscribeOn(Schedulers.parallel()) .subscribe(); .transform(fn) .transform(fn) .transform(fn) .transform(fn)
126. Hooks.onLastOperator(fn) @bsideup Flux.range(0, 5) .map(it -> it.toString()) .single() .subscribeOn(Schedulers.parallel()) .subscribe();
127. Hooks.onLastOperator(fn) @bsideup Flux.range(0, 5) .map(it -> it.toString()) .single() .subscribeOn(Schedulers.parallel()) .transform(fn) .subscribe();
128. Your Reactor should be like:
129. https://9gag.com/gag/aqKWPwY
130. Takeaways • https://github.com/reactor/reactor-tools • Flux#checkpoint() / Flux#log() • Hooks class • https://github.com/reactor/BlockHound • Async APIs - trust, but verify @bsideup
131. @bsideup bsideup