Repository: apache/eventmesh
Issue: [Bug] Pravega connector writer doesn't close when unsubscribing

### Search before asking

- [X] I had searched in the [issues](https://github.com/apache/eventmesh/issues?q=is%3Aissue) and found no similar issues.


### Environment

Windows

### EventMesh version

master

### What happened

Pravega connector writer doesn't close when unsubscribing.

### How to reproduce

Read praveag connector code will find it.

### Debug logs

_No response_

### Are you willing to submit PR?

- [X] Yes I am willing to submit a PR!

Changed files:
eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/client/PravegaEvent.java
eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/SubscribeTask.java
eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/client/PravegaClient.java

diff --git a/eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/client/PravegaClient.java b/eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/client/PravegaClient.java
index 2b7721a60..25ebb53eb 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/client/PravegaClient.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/client/PravegaClient.java
@@ -19,7 +19,6 @@ package org.apache.eventmesh.connector.pravega.client;
 
 import org.apache.eventmesh.api.EventListener;
 import org.apache.eventmesh.api.SendResult;
-import org.apache.eventmesh.connector.pravega.SubscribeTask;
 import org.apache.eventmesh.connector.pravega.config.PravegaConnectorConfig;
 import org.apache.eventmesh.connector.pravega.exception.PravegaConnectorException;
 
@@ -84,7 +83,7 @@ public class PravegaClient {
     }
 
     public void start() {
-        if (PravegaClient.getInstance().createScope()) {
+        if (createScope()) {
             log.info("Create Pravega scope[{}] success.", PravegaConnectorConfig.getInstance().getScope());
         } else {
             log.info("Pravega scope[{}] has already been created.", PravegaConnectorConfig.getInstance().getScope());
@@ -101,6 +100,15 @@ public class PravegaClient {
         streamManager.close();
     }
 
+    /**
+     * Publish CloudEvent to Pravega stream named topic. Note that the messageId in SendResult is always -1
+     * since {@link EventStreamWriter#writeEvent(Object)} just return {@link java.util.concurrent.CompletableFuture}
+     * with {@link Void} which couldn't get messageId.
+     *
+     * @param topic      topic
+     * @param cloudEvent cloudEvent
+     * @return SendResult whose messageId is always -1
+     */
     public SendResult publish(String topic, CloudEvent cloudEvent) {
         if (!createStream(topic)) {
             log.debug("stream[{}] has already been created.", topic);
@@ -147,6 +155,7 @@ public class PravegaClient {
         }
         deleteReaderGroup(buildReaderGroup(topic, consumerGroup));
         subscribeTaskMap.remove(topic).stopRead();
+        writerMap.remove(topic).close();
         return true;
     }
 
diff --git a/eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/client/PravegaEvent.java b/eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/client/PravegaEvent.java
index ec8ac1685..4ce37cc60 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/client/PravegaEvent.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/client/PravegaEvent.java
@@ -36,6 +36,8 @@ import lombok.NoArgsConstructor;
 @Data
 @NoArgsConstructor
 public class PravegaEvent implements Serializable {
+    private static final long serialVersionUID = 0L;
+
     private SpecVersion version;
     private String topic;
     private String data;
diff --git a/eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/SubscribeTask.java b/eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/client/SubscribeTask.java
similarity index 95%
rename from eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/SubscribeTask.java
rename to eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/client/SubscribeTask.java
index 17a73de0c..a38e1159f 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/SubscribeTask.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/client/SubscribeTask.java
@@ -15,12 +15,11 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.connector.pravega;
+package org.apache.eventmesh.connector.pravega.client;
 
 import org.apache.eventmesh.api.EventListener;
 import org.apache.eventmesh.api.EventMeshAction;
 import org.apache.eventmesh.api.EventMeshAsyncConsumeContext;
-import org.apache.eventmesh.connector.pravega.client.PravegaEvent;
 
 import java.util.concurrent.atomic.AtomicBoolean;