Merge branch 'developer' of 47.106.101.75:abel_cjy/Amazon-Selection-Data into developer

Pyspark_job/utils/templates.py

@@ -345,15 +345,23 @@ class Templates(object):
         pass
 
     def kafka_stream_stop(self):
-        try:
-            self.start_process_instance()  # 开启海豚调度
-            if self.query is not None:
-                self.query.stop()  # 退出实时消费（移除awaitTermination避免线程冲突）
-            if self.spark is not None:
-                self.spark.stop()
-            exit(0) # 退出程序
-        except Exception as e:
-            print(e, traceback.format_exc())
+        import threading, os
+        self.start_process_instance()  # 开启海豚调度
+
+        def _do_stop():
+            try:
+                if self.query is not None:
+                    self.query.stop()  # 在子线程中调用，避免 foreachBatch 回调内死锁
+                if self.spark is not None:
+                    self.spark.stop()
+            except Exception as e:
+                print(e, traceback.format_exc())
+            finally:
+                os._exit(0)  # 强制退出 JVM，不被 try/except 拦截
+
+        t = threading.Thread(target=_do_stop, daemon=False)
+        t.start()
+        # foreachBatch 回调从此处正常返回，不阻塞等待 stop 完成
 
     # def kafka_consumption_is_finished(self):
     #     while True: