From 0f1330b72903d7ace49bf907c859d52d8118288e Mon Sep 17 00:00:00 2001
From: Jeff Xiang <31421034+jeffxiang@users.noreply.github.com>
Date: Fri, 23 Feb 2024 14:18:32 -0500
Subject: [PATCH] Fix deserializer.open() NPE by moving invocation to all cases
 (#29)

---
 pom.xml                                                    | 2 +-
 psc-common/pom.xml                                         | 2 +-
 psc-examples/pom.xml                                       | 4 ++--
 psc-flink-logging/pom.xml                                  | 4 ++--
 psc-flink/pom.xml                                          | 2 +-
 .../streaming/connectors/psc/FlinkPscConsumerBase.java     | 7 +++----
 psc-integration-test/pom.xml                               | 2 +-
 psc-logging/pom.xml                                        | 2 +-
 psc/pom.xml                                                | 2 +-
 9 files changed, 13 insertions(+), 14 deletions(-)

diff --git a/pom.xml b/pom.xml
index 4b5f1a2..7181b31 100644
--- a/pom.xml
+++ b/pom.xml
@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.pinterest.psc</groupId>
     <artifactId>psc-java-oss</artifactId>
-    <version>3.1.2</version>
+    <version>3.1.3-RC1</version>
     <packaging>pom</packaging>
     <name>psc-java-oss</name>
     <modules>
diff --git a/psc-common/pom.xml b/psc-common/pom.xml
index b7ac485..82593ec 100644
--- a/psc-common/pom.xml
+++ b/psc-common/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>psc-java-oss</artifactId>
         <groupId>com.pinterest.psc</groupId>
-        <version>3.1.2</version>
+        <version>3.1.3-RC1</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
diff --git a/psc-examples/pom.xml b/psc-examples/pom.xml
index b99a5db..1537e25 100644
--- a/psc-examples/pom.xml
+++ b/psc-examples/pom.xml
@@ -5,13 +5,13 @@
     <parent>
         <artifactId>psc-java-oss</artifactId>
         <groupId>com.pinterest.psc</groupId>
-        <version>3.1.2</version>
+        <version>3.1.3-RC1</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
     <artifactId>psc-examples</artifactId>
-    <version>3.1.2</version>
+    <version>3.1.3-RC1</version>
     <name>psc-examples</name>
 
     <properties>
diff --git a/psc-flink-logging/pom.xml b/psc-flink-logging/pom.xml
index 97265e0..f0f0610 100644
--- a/psc-flink-logging/pom.xml
+++ b/psc-flink-logging/pom.xml
@@ -5,13 +5,13 @@
     <parent>
         <artifactId>psc-java-oss</artifactId>
         <groupId>com.pinterest.psc</groupId>
-        <version>3.1.2</version>
+        <version>3.1.3-RC1</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
     <artifactId>psc-flink-logging</artifactId>
-    <version>3.1.2</version>
+    <version>3.1.3-RC1</version>
 
     <dependencies>
         <dependency>
diff --git a/psc-flink/pom.xml b/psc-flink/pom.xml
index 96e0b22..589903b 100644
--- a/psc-flink/pom.xml
+++ b/psc-flink/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <groupId>com.pinterest.psc</groupId>
         <artifactId>psc-java-oss</artifactId>
-        <version>3.1.2</version>
+        <version>3.1.3-RC1</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <artifactId>psc-flink</artifactId>
diff --git a/psc-flink/src/main/java/com/pinterest/flink/streaming/connectors/psc/FlinkPscConsumerBase.java b/psc-flink/src/main/java/com/pinterest/flink/streaming/connectors/psc/FlinkPscConsumerBase.java
index cc9ab1a..db422ac 100644
--- a/psc-flink/src/main/java/com/pinterest/flink/streaming/connectors/psc/FlinkPscConsumerBase.java
+++ b/psc-flink/src/main/java/com/pinterest/flink/streaming/connectors/psc/FlinkPscConsumerBase.java
@@ -752,11 +752,10 @@ public void open(Configuration configuration) throws Exception {
                 LOG.info("Consumer subtask {} initially has no partitions to read from.",
                         getRuntimeContext().getIndexOfThisSubtask());
             }
-
-            this.deserializer.open(RuntimeContextInitializationContextAdapters.deserializationAdapter(this.getRuntimeContext(), (metricGroup) -> {
-                return metricGroup.addGroup("user");
-            }));
         }
+        this.deserializer.open(RuntimeContextInitializationContextAdapters.deserializationAdapter(this.getRuntimeContext(), (metricGroup) -> {
+            return metricGroup.addGroup("user");
+        }));
     }
 
     @Override
diff --git a/psc-integration-test/pom.xml b/psc-integration-test/pom.xml
index 720f614..42b56f9 100644
--- a/psc-integration-test/pom.xml
+++ b/psc-integration-test/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>psc-java-oss</artifactId>
         <groupId>com.pinterest.psc</groupId>
-        <version>3.1.2</version>
+        <version>3.1.3-RC1</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
diff --git a/psc-logging/pom.xml b/psc-logging/pom.xml
index 4ee93b1..d87b51f 100644
--- a/psc-logging/pom.xml
+++ b/psc-logging/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>psc-java-oss</artifactId>
         <groupId>com.pinterest.psc</groupId>
-        <version>3.1.2</version>
+        <version>3.1.3-RC1</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
diff --git a/psc/pom.xml b/psc/pom.xml
index bdad2bc..97b7332 100644
--- a/psc/pom.xml
+++ b/psc/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>psc-java-oss</artifactId>
         <groupId>com.pinterest.psc</groupId>
-        <version>3.1.2</version>
+        <version>3.1.3-RC1</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>