[flink] branch release-1.14 updated: [FLINK-24206][connector/pulsar] Close the pulsar client properly.

[xt...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

xtsong pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.14 by this push:
     new 28677f3  [FLINK-24206][connector/pulsar] Close the pulsar client properly.
28677f3 is described below

commit 28677f34bad1ffb4f405a4550536969eb9621d05
Author: syhily <sy...@gmail.com>
AuthorDate: Mon Sep 13 16:40:08 2021 +0800

    [FLINK-24206][connector/pulsar] Close the pulsar client properly.
    
    Pulsar Consumer would be closed in SplitReader if it reaches the end of split. The registered Consumer would also be closed in PulsarClient if we call PulsarClient.close(). This would cause race condition. We have to use PulsarClient.shutdown() which don't close Consumer instead.
    
    This closes #17255
---
 .../pulsar/source/reader/source/PulsarSourceReaderBase.java        | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarSourceReaderBase.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarSourceReaderBase.java
index d340b9f..2792895 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarSourceReaderBase.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarSourceReaderBase.java
@@ -79,10 +79,11 @@ abstract class PulsarSourceReaderBase<OUT>
 
     @Override
     public void close() throws Exception {
+        // Close the all the consumers first.
+        super.close();
+
         // Close shared pulsar resources.
-        pulsarClient.close();
+        pulsarClient.shutdown();
         pulsarAdmin.close();
-
-        super.close();
     }
 }