diff --git a/skywalking/agent/protocol/grpc.py b/skywalking/agent/protocol/grpc.py index f5468a44..42070702 100644 --- a/skywalking/agent/protocol/grpc.py +++ b/skywalking/agent/protocol/grpc.py @@ -18,7 +18,7 @@ import logging from skywalking.loggings import logger import traceback -from queue import Queue +from queue import Queue, Empty import grpc @@ -70,7 +70,10 @@ def on_error(self): def report(self, queue: Queue, block: bool = True): def generator(): while True: - segment = queue.get(block=block) # type: Segment + try: + segment = queue.get(block=block) # type: Segment + except Empty: + return logger.debug('reporting segment %s', segment) diff --git a/skywalking/agent/protocol/http.py b/skywalking/agent/protocol/http.py index 331f71a9..89d43bfb 100644 --- a/skywalking/agent/protocol/http.py +++ b/skywalking/agent/protocol/http.py @@ -16,7 +16,7 @@ # from skywalking.loggings import logger -from queue import Queue +from queue import Queue, Empty from skywalking.agent import Protocol from skywalking.client.http import HttpServiceManagementClient, HttpTraceSegmentReportService @@ -41,7 +41,10 @@ def connected(self): def report(self, queue: Queue, block: bool = True): def generator(): while True: - segment = queue.get(block=block) # type: Segment + try: + segment = queue.get(block=block) # type: Segment + except Empty: + return logger.debug('reporting segment %s', segment) diff --git a/skywalking/agent/protocol/kafka.py b/skywalking/agent/protocol/kafka.py index 8e6e75af..3eabfdbb 100644 --- a/skywalking/agent/protocol/kafka.py +++ b/skywalking/agent/protocol/kafka.py @@ -17,7 +17,7 @@ import logging from skywalking.loggings import logger, getLogger -from queue import Queue +from queue import Queue, Empty from skywalking import config from skywalking.agent import Protocol @@ -45,7 +45,10 @@ def heartbeat(self): def report(self, queue: Queue, block: bool = True): def generator(): while True: - segment = queue.get(block=block) # type: Segment + try: + segment = queue.get(block=block) # type: Segment + except Empty: + return logger.debug('reporting segment %s', segment)