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

1. Don’t be Homer Simpson 
 with your Reactor! Sergei Egorov, Pivotal @bsideup https://simpsonswiki.com/wiki/File:Springfield_Nuclear_Power_Plant_6.png
2. • Staff Engineer at Pivotal • Berlin Spring User Group co-organizer • Testcontainers co-maintainer About me @bsideup
5. Remember Peter…
6. With great performance comes… Remember Peter…
7. With great performance comes…
 great PITA*. Remember Peter… * PITA - performance improvement to apps
8. My first experience with
 reactive programming was like…
9. @bsideup
10. @bsideup
11. Mono.fromRunnable()
12. Mono.fromRunnable() @bsideup
13. Mono.fromRunnable() My code @bsideup
14. Mono.fromRunnable() My code My code @bsideup
15. Mono.fromRunnable() My code My code My code Work! @bsideup
16. Mono.fromRunnable().subscribe() @bsideup
17. 1 week later…
18. 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
19. 1 week of production later…
20. @bsideup
21. What could go wrong
22. Whoa #1: exceptions
23. // com/example/demo/Example.java Flux.range(0, 5) .single() .subscribeOn(Schedulers.parallel()) .subscribe(); @bsideup
24. // 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
25. // 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
26. // 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
27. // 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
28. // 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
29. Assembly vs Subscription
30. Flux.range(0, 5) Assembly @bsideup
31. Flux.range(0, 5) .filter(it -> it > 3) Assembly @bsideup
32. Flux.range(0, 5) .filter(it -> it > 3) .map(it -> it.toString()) Assembly @bsideup
33. Flux.range(0, 5) .filter(it -> it > 3) .map(it -> it.toString()) .single() Assembly @bsideup
34. Flux.range(0, 5) .filter(it -> it > 3) .map(it -> it.toString()) .single() .subscribeOn(Schedulers.parallel()) Assembly @bsideup
35. Flux.range(0, 5) .filter(it -> it > 3) .map(it -> it.toString()) .single() .subscribeOn(Schedulers.parallel()) .subscribe() Assembly @bsideup
36. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 @bsideup
37. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 * = subscriber @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 .onSubscribe( ) * = subscription @bsideup
43. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 @bsideup
44. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 @bsideup
45. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 @bsideup
46. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 @bsideup
47. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 .request(max) @bsideup
48. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 .request(max) @bsideup
49. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 .request(max) @bsideup
50. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 .request(max) @bsideup
51. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 .request(max) @bsideup
52. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 .request(max) @bsideup
53. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Execution current thread parallel-1 .onNext(0) @bsideup
54. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Execution current thread parallel-1 .onNext(0) @bsideup
55. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Execution current thread parallel-1 .onNext(0) @bsideup
56. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Execution current thread parallel-1 .onNext(0) @bsideup
57. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Execution current thread parallel-1 .onNext(0) @bsideup
58. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Execution current thread parallel-1 .onNext(1) @bsideup
59. FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Execution current thread parallel-1 .onNext(1) @bsideup
60. 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
61. Demo http://eskipaper.com/homer-jay-simpson-cartoon.html
62. How ReactorDebugAgent works? @bsideup
63. // com/example/demo/Example.java Flux.range(0, 5) .single() .subscribeOn(Schedulers.parallel()) .subscribe(); @bsideup
64. // com/example/demo/Example.java Flux.range(0, 5) .single() .subscribeOn(Schedulers.parallel()) .subscribe(); @bsideup
65. // 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
66. 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
67. Whoa #2: Blocking calls! https://blog.tfd.co.uk/2010/10/15/jackrabbit-performance/ @bsideup
68. How Reactor schedules tasks @bsideup
69. 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
70. Mono.delay(ofSeconds(1)) @bsideup
71. Mono.delay(ofSeconds(1)) Mono.delay(ofSeconds(1), Schedulers.parallel()) @bsideup
72. Mono.delay(ofSeconds(1)) Mono.delay(ofSeconds(1), Schedulers.parallel()) Every non-instant operation 
 runs on the parallel scheduler
 by default! @bsideup
73. parallel-1 parallel-2 parallel-3 parallel-4 Non-blocking execution @bsideup
74. parallel-1 parallel-2 parallel-3 parallel-4 Non-blocking execution @bsideup
75. parallel-1 parallel-2 parallel-3 parallel-4 Non-blocking execution @bsideup
76. parallel-1 parallel-2 parallel-3 parallel-4 Non-blocking execution @bsideup
77. parallel-1 parallel-2 parallel-3 parallel-4 Non-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 @bsideup
83. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution @bsideup
84. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution @bsideup
85. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution @bsideup
86. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution @bsideup
87. parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution ⚠ no more tasks scheduled 
 until this task returns @bsideup
88. How to fix?
89. Project Loom
90. @bsideup
91. @bsideup
92. Use non-blocking APIs, or…
93. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
94. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
95. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
96. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
97. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
98. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
99. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
100. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
101. parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup
102. Demo http://eskipaper.com/homer-jay-simpson-cartoon.html
103. How BlockHound works? @bsideup
104. public class Thread implements Runnable { public static native void sleep(long millis); } @bsideup
105. public class Thread implements Runnable { private static native void $$BlockHound$$_sleep(long millis); public static void sleep(long millis) { $$BlockHound$$_sleep(millis); } } @bsideup
106. 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
107. 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
108. Blocking check @bsideup
109. Blocking check 1. Get or tag the current thread • JVMTI’s built-in mechanism to tag objects, calls nonBlockingThreadPredicate on creation and caches the result forever @bsideup
110. Blocking check 1. Get or tag the 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
111. Blocking check 1. Get or tag the 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
112. 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)
113. 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)
114. 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)
115. 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)
116. 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");
117. Stack @bsideup at java.io.FileOutputStream.writeBytes(FileOutputStream.java)
118. Stack @bsideup at java.io.FileOutputStream.writeBytes(FileOutputStream.java) at java.io.FileOutputStream.write(FileOutputStream.java:326) at java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:82) at java.io.BufferedOutputStream.flush(BufferedOutputStream.java:140) at java.io.PrintStream.write(PrintStream.java:482) at java.io.FilterOutputStream.write(FilterOutputStream.java:97)
119. Stack @bsideup at java.io.FileOutputStream.writeBytes(FileOutputStream.java) at java.io.FileOutputStream.write(FileOutputStream.java:326) at java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:82) at java.io.BufferedOutputStream.flush(BufferedOutputStream.java:140) at java.io.PrintStream.write(PrintStream.java:482) at java.io.FilterOutputStream.write(FilterOutputStream.java:97) at ch.qos.logback.core.joran.spi.ConsoleTarget$1.write(ConsoleTarget.java:37) at ch.qos.logback.core.OutputStreamAppender.writeBytes(OutputStreamAppender.java:199) at ch.qos.logback.core.OutputStreamAppender.subAppend(OutputStreamAppender.java:231) at ch.qos.logback.core.OutputStreamAppender.append(OutputStreamAppender.java:102) at ch.qos.logback.core.UnsynchronizedAppenderBase.doAppend(UnsynchronizedAppenderBase.java:84) at ch.qos.logback.core.spi.AppenderAttachableImpl.appendLoopOnAppenders(AppenderAttachableImpl.java:51) at ch.qos.logback.classic.Logger.appendLoopOnAppenders(Logger.java:270) at ch.qos.logback.classic.Logger.callAppenders(Logger.java:257) builder.allowBlockingCallsInside( “ch.qos.logback.classic.Logger", “callAppenders" );
120. 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
121. “I use async APIs and I am safe!”
122. Yeah… sure.
123. Or…
124. Are you sure?
125. KafkaProducer#send(ProducerRecord,Callback) @bsideup
126. KafkaProducer#send(ProducerRecord,Callback) “Asynchronously send a record to a topic and invoke the provided callback when the send has been acknowledged.” - Javadoc @bsideup
127. 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
128. 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
129. 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
130. @bsideup
131. How to instrument Reactor
132. Schedulers.onScheduleHook(fn) @bsideup
133. Schedulers.onScheduleHook(fn) @bsideup Schedulers.onScheduleHook("myHook", runnable -> { println("Before every scheduled runnable"); return () -> { println("Before execution"); runnable.run(); println("After execution"); }; });
134. Schedulers.onScheduleHook(fn) @bsideup Schedulers.onScheduleHook("mdc", runnable -> { String userId = MDC.get("userId"); return () -> { MDC.put("userId", userId); try { runnable.run(); } finally { MDC.remove("userId"); } }; });
135. Hooks.onEachOperator(fn) @bsideup Flux.range(0, 5) .map(it -> it.toString()) .single() .subscribeOn(Schedulers.parallel()) .subscribe();
136. Hooks.onEachOperator(fn) @bsideup Flux.range(0, 5) .map(it -> it.toString()) .single() .subscribeOn(Schedulers.parallel()) .subscribe();
137. 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)
138. Hooks.onLastOperator(fn) @bsideup Flux.range(0, 5) .map(it -> it.toString()) .single() .subscribeOn(Schedulers.parallel()) .subscribe();
139. Hooks.onLastOperator(fn) @bsideup Flux.range(0, 5) .map(it -> it.toString()) .single() .subscribeOn(Schedulers.parallel()) .transform(fn) .subscribe();
140. Your Reactor should be like:
141. https://9gag.com/gag/aqKWPwY
142. Takeaways • https://github.com/reactor/reactor-tools • Flux#checkpoint() / Flux#log() • Hooks class • https://github.com/reactor/BlockHound • Async APIs - trust, but verify @bsideup
143. @bsideup bsideup