[ISSUE #4264]Some optimizations for ProducerService (#4334)

847850277 · web-flow · commit 800b628af50b · 2023-08-05T20:27:31.000+08:00
diff --git a/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/service/ConsumerService.java b/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/service/ConsumerService.java
@@ -42,11 +42,11 @@
 @Slf4j
 public class ConsumerService extends ConsumerServiceGrpc.ConsumerServiceImplBase {
 
-    public OpenFunctionSinkConnector openFunctionSinkConnector;
+    private final OpenFunctionSinkConnector openFunctionSinkConnector;
 
-    public BlockingQueue<ConnectRecord> queue;
+    private final BlockingQueue<ConnectRecord> queue;
 
-    public OpenFunctionServerConfig config;
+    private final OpenFunctionServerConfig config;
 
     private final transient ManagedChannel channel = ManagedChannelBuilder.forAddress("127.0.0.1", 10115).usePlaintext().build();
 
diff --git a/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/service/ProducerService.java b/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/service/ProducerService.java
@@ -44,11 +44,11 @@
 @Slf4j
 public class ProducerService extends PublisherServiceGrpc.PublisherServiceImplBase {
 
-    public OpenFunctionSourceConnector openFunctionSourceConnector;
+    private final OpenFunctionSourceConnector openFunctionSourceConnector;
 
-    public BlockingQueue<ConnectRecord> queue;
+    private final BlockingQueue<ConnectRecord> queue;
 
-    public OpenFunctionServerConfig config;
+    private final OpenFunctionServerConfig config;
 
     public ProducerService(OpenFunctionSourceConnector openFunctionSourceConnector, OpenFunctionServerConfig serverConfig) {
         this.openFunctionSourceConnector = openFunctionSourceConnector;
@@ -85,7 +85,7 @@ public void publish(CloudEvent event, StreamObserver<CloudEvent> responseObserve
                     CloudEventAttributeValue.newBuilder().setCeString(StatusCode.EVENTMESH_SEND_ASYNC_MSG_ERR.getErrMsg()).build())
                 .putAttributes(ProtocolKey.GRPC_RESPONSE_TIME, CloudEventAttributeValue.newBuilder()
                     .setCeTimestamp(Timestamp.newBuilder().setSeconds(instant.getEpochSecond()).setNanos(instant.getNano()).build()).build());
-            throw new RuntimeException(e);
+            Thread.currentThread().interrupt();
         }
 
         responseObserver.onNext(builder.build());
