[skywalking] branch master updated: If transform throw exception, the stream observer won't call the onCompleted method, then the stream observer will wait it finish until the timeout. (#3163)

[wusheng]

This is an automated email from the ASF dual-hosted git repository.

wusheng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/skywalking.git


The following commit(s) were added to refs/heads/master by this push:
     new 0cf9980  If transform throw exception, the stream observer won't call 
the onCompleted method, then the stream observer will wait it finish until the 
timeout. (#3163)
0cf9980 is described below

commit 0cf998022de52d9e23b4ce3010e3a0503eb33a41
Author: 彭勇升 pengys <pen...@apache.org>
AuthorDate: Thu Jul 25 08:54:14 2019 +0800

    If transform throw exception, the stream observer won't call the 
onCompleted method, then the stream observer will wait it finish until the 
timeout. (#3163)
---
 .../apm/agent/core/remote/TraceSegmentServiceClient.java         | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

diff --git 
a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/TraceSegmentServiceClient.java
 
b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/TraceSegmentServiceClient.java
index a58f6c8..1645746 100644
--- 
a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/TraceSegmentServiceClient.java
+++ 
b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/TraceSegmentServiceClient.java
@@ -113,13 +113,14 @@ public class TraceSegmentServiceClient implements 
BootService, IConsumer<TraceSe
                     UpstreamSegment upstreamSegment = segment.transform();
                     upstreamSegmentStreamObserver.onNext(upstreamSegment);
                 }
-                upstreamSegmentStreamObserver.onCompleted();
-
-                status.wait4Finish();
-                segmentUplinkedCounter += data.size();
             } catch (Throwable t) {
                 logger.error(t, "Transform and send UpstreamSegment to 
collector fail.");
             }
+
+            upstreamSegmentStreamObserver.onCompleted();
+
+            status.wait4Finish();
+            segmentUplinkedCounter += data.size();
         } else {
             segmentAbandonedCounter += data.size();
         }