SinkClient fix logging on unsubscribe (#144)

* Add logging to SinkClient to track uncleaned connections * Only count resubmission from worker failures * SinkClient fix logging on unsubscribe Co-authored-by: Calvin Cheung <[email protected]>
Netflix · Mar 13, 2022 · 9f67db4 · 9f67db4
1 parent de88e88
commit 9f67db4
Showing 1 changed file with 3 additions and 3 deletions.
diff --git a/mantis-client/src/main/java/io/mantisrx/client/SinkClientImpl.java b/mantis-client/src/main/java/io/mantisrx/client/SinkClientImpl.java
@@ -144,10 +144,10 @@ public Observable<T> call(EndpointChange endpointChange) {
                 })
                 .doOnUnsubscribe(() -> {
                     try {
-                        logger.warn("Closing connections to sink of job " + jobId);
+                        logger.warn("Closing connections to sink of job {}", jobId);
                         closeAllConnections();
                     } catch (Exception e) {
-                        Observable.error(e);
+                        logger.warn("Error closing all connections to sink of job {}", jobId, e);
                     }
                 })
                 .share()
@@ -212,7 +212,7 @@ public void call(Boolean flag) {
             }
         }
         return ((SinkConnection<T>) sinkConnection).call()
-                //.flatMap(o -> o)
+                .takeWhile(e -> !nowClosed.get())
                 ;
     }