Fix missed events in EventHandlers when flatMapping.

joshiste · joshiste · commit 452b618f49b9 · 2019-03-23T23:32:10.000+01:00
An error during event handling for a single event doesn't kill the sub- scription, but prevents the remaining events of the current "chunk" to be processed, so we need to ignore failures for single events. closes #1128
diff --git a/spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/services/EndpointDetectionTrigger.java b/spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/services/EndpointDetectionTrigger.java
@@ -25,8 +25,11 @@
 import reactor.core.scheduler.Schedulers;
 
 import org.reactivestreams.Publisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class EndpointDetectionTrigger extends AbstractEventHandler<InstanceEvent> {
+    private static final Logger log = LoggerFactory.getLogger(EndpointDetectionTrigger.class);
     private final EndpointDetector endpointDetector;
 
     public EndpointDetectionTrigger(EndpointDetector endpointDetector, Publisher<InstanceEvent> publisher) {
@@ -45,6 +48,9 @@ protected Publisher<Void> handle(Flux<InstanceEvent> publisher) {
     }
 
     protected Mono<Void> detectEndpoints(InstanceEvent event) {
-        return endpointDetector.detectEndpoints(event.getInstance());
+        return this.endpointDetector.detectEndpoints(event.getInstance()).onErrorResume(e -> {
+            log.warn("Unexpected error while detecting endpoints for {}", event.getInstance(), e);
+            return Mono.empty();
+        });
     }
 }
diff --git a/spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/services/InfoUpdateTrigger.java b/spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/services/InfoUpdateTrigger.java
@@ -28,8 +28,11 @@
 
 import java.time.Duration;
 import org.reactivestreams.Publisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class InfoUpdateTrigger extends AbstractEventHandler<InstanceEvent> {
+    private static final Logger log = LoggerFactory.getLogger(InfoUpdateTrigger.class);
     private final InfoUpdater infoUpdater;
     private final IntervalCheck intervalCheck;
 
@@ -51,7 +54,10 @@ protected Publisher<Void> handle(Flux<InstanceEvent> publisher) {
     }
 
     protected Mono<Void> updateInfo(InstanceId instanceId) {
-        return this.infoUpdater.updateInfo(instanceId).doFinally(s -> this.intervalCheck.markAsChecked(instanceId));
+        return this.infoUpdater.updateInfo(instanceId).onErrorResume(e -> {
+            log.warn("Unexpected error while updating info for {}", instanceId, e);
+            return Mono.empty();
+        }).doFinally(s -> this.intervalCheck.markAsChecked(instanceId));
     }
 
     @Override
diff --git a/spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/services/StatusUpdateTrigger.java b/spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/services/StatusUpdateTrigger.java
@@ -27,8 +27,11 @@
 
 import java.time.Duration;
 import org.reactivestreams.Publisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class StatusUpdateTrigger extends AbstractEventHandler<InstanceEvent> {
+    private static final Logger log = LoggerFactory.getLogger(StatusUpdateTrigger.class);
     private final StatusUpdater statusUpdater;
     private final IntervalCheck intervalCheck;
 
@@ -49,7 +52,10 @@ protected Publisher<Void> handle(Flux<InstanceEvent> publisher) {
     }
 
     protected Mono<Void> updateStatus(InstanceId instanceId) {
-        return this.statusUpdater.updateStatus(instanceId).doFinally(s -> this.intervalCheck.markAsChecked(instanceId));
+        return this.statusUpdater.updateStatus(instanceId).onErrorResume(e -> {
+            log.warn("Unexpected error while updating status for {}", instanceId, e);
+            return Mono.empty();
+        }).doFinally(s -> this.intervalCheck.markAsChecked(instanceId));
     }
 
     @Override
diff --git a/spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/services/AbstractEventHandlerTest.java b/spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/services/AbstractEventHandlerTest.java
@@ -102,17 +102,17 @@ private TestEventHandler(Publisher<InstanceEvent> publisher) {
         protected Publisher<Void> handle(Flux<InstanceRegisteredEvent> publisher) {
             return publisher.flatMap(event -> {
                 if (event.equals(errorEvent)) {
-                    throw (new IllegalStateException("Error"));
+                    return Mono.error(new IllegalStateException("Error"));
                 } else {
                     log.info("Event {}", event);
-                    sink.next(event);
+                    this.sink.next(event);
                     return Mono.empty();
                 }
-            });
+            }).then();
         }
 
         public Flux<InstanceEvent> getFlux() {
-            return flux;
+            return this.flux;
         }
     }
 
diff --git a/spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/services/EndpointDetectionTriggerTest.java b/spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/services/EndpointDetectionTriggerTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2018 the original author or authors.
+ * Copyright 2014-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -47,44 +47,74 @@ public class EndpointDetectionTriggerTest {
 
     @Before
     public void setUp() throws Exception {
-        when(detector.detectEndpoints(any(InstanceId.class))).thenReturn(Mono.empty());
-        trigger = new EndpointDetectionTrigger(detector, events.flux());
-        trigger.start();
+        when(this.detector.detectEndpoints(any(InstanceId.class))).thenReturn(Mono.empty());
+        this.trigger = new EndpointDetectionTrigger(this.detector, this.events.flux());
+        this.trigger.start();
         Thread.sleep(50L); //wait for subscription
     }
 
     @Test
     public void should_detect_on_status_changed() {
         //when status-change event is emitted
-        events.next(new InstanceStatusChangedEvent(instance.getId(), instance.getVersion(), StatusInfo.ofDown()));
+        this.events.next(new InstanceStatusChangedEvent(this.instance.getId(),
+            this.instance.getVersion(),
+            StatusInfo.ofDown()
+        ));
         //then should update
-        verify(detector, times(1)).detectEndpoints(instance.getId());
+        verify(this.detector, times(1)).detectEndpoints(this.instance.getId());
     }
 
     @Test
     public void should_detect_on_registration_updated() {
         //when status-change event is emitted
-        events.next(
-            new InstanceRegistrationUpdatedEvent(instance.getId(), instance.getVersion(), instance.getRegistration()));
+        this.events.next(new InstanceRegistrationUpdatedEvent(this.instance.getId(),
+            this.instance.getVersion(),
+            this.instance.getRegistration()
+        ));
         //then should update
-        verify(detector, times(1)).detectEndpoints(instance.getId());
+        verify(this.detector, times(1)).detectEndpoints(this.instance.getId());
     }
 
     @Test
     public void should_not_detect_on_non_relevant_event() {
         //when some non-status-change event is emitted
-        events.next(new InstanceRegisteredEvent(instance.getId(), instance.getVersion(), instance.getRegistration()));
+        this.events.next(new InstanceRegisteredEvent(this.instance.getId(),
+            this.instance.getVersion(),
+            this.instance.getRegistration()
+        ));
         //then should not update
-        verify(detector, never()).detectEndpoints(instance.getId());
+        verify(this.detector, never()).detectEndpoints(this.instance.getId());
     }
 
     @Test
     public void should_not_detect_on_trigger_stopped() {
         //when registered event is emitted but the trigger has been stopped
-        trigger.stop();
-        clearInvocations(detector);
-        events.next(new InstanceRegisteredEvent(instance.getId(), instance.getVersion(), instance.getRegistration()));
+        this.trigger.stop();
+        clearInvocations(this.detector);
+        this.events.next(new InstanceRegisteredEvent(this.instance.getId(),
+            this.instance.getVersion(),
+            this.instance.getRegistration()
+        ));
         //then should not update
-        verify(detector, never()).detectEndpoints(instance.getId());
+        verify(this.detector, never()).detectEndpoints(this.instance.getId());
+    }
+
+    @Test
+    public void should_continue_detection_after_error() throws InterruptedException {
+        //when status-change event is emitted and an error is emitted
+        when(this.detector.detectEndpoints(any())).thenReturn(Mono.error(IllegalStateException::new))
+                                                  .thenReturn(Mono.empty());
+
+        this.events.next(new InstanceStatusChangedEvent(this.instance.getId(),
+            this.instance.getVersion(),
+            StatusInfo.ofDown()
+        ));
+        this.events.next(new InstanceStatusChangedEvent(this.instance.getId(),
+            this.instance.getVersion(),
+            StatusInfo.ofUp()
+        ));
+
+        //then should update
+        verify(this.detector, times(2)).detectEndpoints(this.instance.getId());
     }
 }
diff --git a/spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/services/InfoUpdateTriggerTest.java b/spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/services/InfoUpdateTriggerTest.java
@@ -149,4 +149,23 @@ public void should_not_update_on_non_relevant_event() {
         //then should not update
         verify(this.updater, never()).updateInfo(this.instance.getId());
     }
+
+    @Test
+    public void should_continue_update_after_error() throws InterruptedException {
+        //when status-change event is emitted and an error is emitted
+        when(this.updater.updateInfo(any())).thenReturn(Mono.error(IllegalStateException::new))
+                                            .thenReturn(Mono.empty());
+
+        this.events.next(new InstanceStatusChangedEvent(this.instance.getId(),
+            this.instance.getVersion(),
+            StatusInfo.ofDown()
+        ));
+        this.events.next(new InstanceStatusChangedEvent(this.instance.getId(),
+            this.instance.getVersion(),
+            StatusInfo.ofUp()
+        ));
+
+        //then should update
+        verify(this.updater, times(2)).updateInfo(this.instance.getId());
+    }
 }
diff --git a/spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/services/StatusUpdateTriggerTest.java b/spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/services/StatusUpdateTriggerTest.java
@@ -132,4 +132,23 @@ public void should_not_update_on_non_relevant_event() {
         //then should not update
         verify(this.updater, never()).updateStatus(this.instance.getId());
     }
+
+    @Test
+    public void should_continue_update_after_error() throws InterruptedException {
+        //when status-change event is emitted and an error is emitted
+        when(this.updater.updateStatus(any())).thenReturn(Mono.error(IllegalStateException::new))
+                                              .thenReturn(Mono.empty());
+
+        this.events.next(new InstanceRegistrationUpdatedEvent(this.instance.getId(),
+            this.instance.getVersion(),
+            this.instance.getRegistration()
+        ));
+        this.events.next(new InstanceRegistrationUpdatedEvent(this.instance.getId(),
+            this.instance.getVersion(),
+            this.instance.getRegistration()
+        ));
+
+        //then should update
+        verify(this.updater, times(2)).updateStatus(this.instance.getId());
+    }
 }

Original file line number	Diff line number	Diff line change
`@@ -25,8 +25,11 @@`
`25`	`25`	`import reactor.core.scheduler.Schedulers;`
`26`	`26`
`27`	`27`	`import org.reactivestreams.Publisher;`
	`28`	`+import org.slf4j.Logger;`
	`29`	`+import org.slf4j.LoggerFactory;`
`28`	`30`
`29`	`31`	`public class EndpointDetectionTrigger extends AbstractEventHandler<InstanceEvent> {`
	`32`	`+ private static final Logger log = LoggerFactory.getLogger(EndpointDetectionTrigger.class);`
`30`	`33`	`private final EndpointDetector endpointDetector;`
`31`	`34`
`32`	`35`	`public EndpointDetectionTrigger(EndpointDetector endpointDetector, Publisher<InstanceEvent> publisher) {`
`@@ -45,6 +48,9 @@ protected Publisher<Void> handle(Flux<InstanceEvent> publisher) {`
`45`	`48`	`}`
`46`	`49`
`47`	`50`	`protected Mono<Void> detectEndpoints(InstanceEvent event) {`
`48`		`- return endpointDetector.detectEndpoints(event.getInstance());`
	`51`	`+ return this.endpointDetector.detectEndpoints(event.getInstance()).onErrorResume(e -> {`
	`52`	`+ log.warn("Unexpected error while detecting endpoints for {}", event.getInstance(), e);`
	`53`	`+ return Mono.empty();`
	`54`	`+ });`
`49`	`55`	`}`
`50`	`56`	`}`
Original file line number	Diff line number	Diff line change
`@@ -102,17 +102,17 @@ private TestEventHandler(Publisher<InstanceEvent> publisher) {`
`102`	`102`	`protected Publisher<Void> handle(Flux<InstanceRegisteredEvent> publisher) {`
`103`	`103`	`return publisher.flatMap(event -> {`
`104`	`104`	`if (event.equals(errorEvent)) {`
`105`		`- throw (new IllegalStateException("Error"));`
	`105`	`+ return Mono.error(new IllegalStateException("Error"));`
`106`	`106`	`} else {`
`107`	`107`	`log.info("Event {}", event);`
`108`		`- sink.next(event);`
	`108`	`+ this.sink.next(event);`
`109`	`109`	`return Mono.empty();`
`110`	`110`	`}`
`111`		`- });`
	`111`	`+ }).then();`
`112`	`112`	`}`
`113`	`113`
`114`	`114`	`public Flux<InstanceEvent> getFlux() {`
`115`		`- return flux;`
	`115`	`+ return this.flux;`
`116`	`116`	`}`
`117`	`117`	`}`
`118`	`118`