Close reader for metadata topic during initialization (apache#8637)

Co-authored-by: Jerry Peng <[email protected]>
eugene-cheverda · Nov 20, 2020 · 95755fd · 95755fd
1 parent a8cd908
commit 95755fd
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 10 deletions.
diff --git a/...ions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionMetaDataManager.java b/...ions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionMetaDataManager.java
@@ -97,9 +97,9 @@ public FunctionMetaDataManager(WorkerConfig workerConfig,
      * We create a new reader
      */
     public synchronized void initialize() {
-        try {
+        try (Reader reader = FunctionMetaDataTopicTailer.createReader(
+          workerConfig, pulsarClient.newReader(), MessageId.earliest)){
             // read all existing messages
-            Reader reader = FunctionMetaDataTopicTailer.createReader(workerConfig, pulsarClient.newReader(), MessageId.earliest);
             while (reader.hasMessageAvailable()) {
                 processMetaDataTopicMessage(reader.readNext());
             }

diff --git a/...tions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java b/...tions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
@@ -227,12 +227,11 @@ public FunctionRuntimeManager(WorkerConfig workerConfig, WorkerService workerSer
      * @return the message id of the message processed during init phase
      */
     public MessageId initialize() {
-        try {
-            Reader<byte[]> reader = WorkerUtils.createReader(
-                    workerService.getClient().newReader(),
-                    workerConfig.getWorkerId() + "-function-assignment-initialize",
-                    workerConfig.getFunctionAssignmentTopic(),
-                    MessageId.earliest);
+        try (Reader<byte[]> reader = WorkerUtils.createReader (
+          workerService.getClient().newReader(),
+          workerConfig.getWorkerId() + "-function-assignment-initialize",
+          workerConfig.getFunctionAssignmentTopic(),
+          MessageId.earliest)) {
 
             // start init phase
             this.isInitializePhase = true;
@@ -246,8 +245,6 @@ public MessageId initialize() {
             }
             // init phase is done
             this.isInitializePhase = false;
-            // close reader
-            reader.close();
             // realize existing assignments
             Map<String, Assignment> assignmentMap = workerIdToAssignments.get(this.workerConfig.getWorkerId());
             if (assignmentMap != null) {