diff --git a/pom.xml b/pom.xml
index 17ab9118..c4a41c04 100644
--- a/pom.xml
+++ b/pom.xml
@@ -30,7 +30,7 @@ under the License.
 
 	<groupId>io.streamnative.connectors</groupId>
 	<artifactId>pulsar-flink-parent</artifactId>
-	<version>1.13.1.0</version>
+	<version>1.13.3.0</version>
 
 	<name>StreamNative :: Pulsar Flink Connector :: Root</name>
 	<packaging>pom</packaging>
@@ -91,8 +91,9 @@ under the License.
 		<slf4j.version>1.7.15</slf4j.version>
 
 		<!-- Protobuf support -->
-		<protobuf.version>3.11.4</protobuf.version>
+		<protobuf.version>3.16.1</protobuf.version>
 		<flink-protobuf.version>2.7.6</flink-protobuf.version>
+		<protobuf-plugin.version>3.11.4</protobuf-plugin.version>
 
 		<!-- Default scala versions, must be overwritten by build profiles, so we set something
 		invalid here -->
diff --git a/pulsar-flink-connector/pom.xml b/pulsar-flink-connector/pom.xml
index fe50e16b..d51cad09 100644
--- a/pulsar-flink-connector/pom.xml
+++ b/pulsar-flink-connector/pom.xml
@@ -234,7 +234,7 @@ under the License.
 			<plugin>
 				<groupId>com.github.os72</groupId>
 				<artifactId>protoc-jar-maven-plugin</artifactId>
-				<version>${protobuf.version}</version>
+				<version>${protobuf-plugin.version}</version>
 				<executions>
 					<execution>
 						<id>generate-protobuf-test-sources</id>
diff --git a/pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/internal/DateTimeUtils.java b/pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/internal/DateTimeUtils.java
index 2c26e104..96edde9b 100644
--- a/pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/internal/DateTimeUtils.java
+++ b/pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/internal/DateTimeUtils.java
@@ -25,6 +25,9 @@
 import java.sql.Timestamp;
 import java.text.DateFormat;
 import java.text.SimpleDateFormat;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
 import java.util.Calendar;
 import java.util.Date;
 import java.util.Locale;
@@ -63,7 +66,8 @@ public static Date stringToTime(String s) {
                 return java.sql.Date.valueOf(s);
             }
         } else {
-            return DatatypeConverter.parseDateTime(s).getTime();
+            LocalDateTime ldt = LocalDateTime.parse(s, DateTimeFormatter.ISO_DATE_TIME);
+            return Date.from(ldt.toInstant(ZoneOffset.UTC));
         }
     }