[Bug Fix] http sink NEP checkpoint restore issue

CHANGELOG.md

@@ -1,6 +1,8 @@
 # Changelog
 
 ## [Unreleased]
+### Fixed
+- Fixed http sink NEP issue when flink job restore from checkpoint.
 
 ## [0.23.0] - 2025-11-07
 

src/main/java/com/getindata/connectors/http/internal/sink/HttpSinkInternal.java

@@ -117,11 +117,7 @@ protected HttpSinkInternal(
     public StatefulSinkWriter<InputT, BufferedRequestState<HttpSinkRequestEntry>> createWriter(
             InitContext context) throws IOException {
 
-        ElementConverter<InputT, HttpSinkRequestEntry> elementConverter = getElementConverter();
-        if (elementConverter instanceof SchemaLifecycleAwareElementConverter) {
-            // This cast is needed for Flink 1.15.3 build
-            ((SchemaLifecycleAwareElementConverter<?, ?>) elementConverter).open(context);
-        }
+        ElementConverter<InputT, HttpSinkRequestEntry> elementConverter = initElementConverterOfSchema(context);
 
         return new HttpSinkWriter<>(
             elementConverter,
@@ -144,12 +140,22 @@ public StatefulSinkWriter<InputT, BufferedRequestState<HttpSinkRequestEntry>> cr
         );
     }
 
+    private ElementConverter<InputT, HttpSinkRequestEntry> initElementConverterOfSchema(InitContext context) {
+        ElementConverter<InputT, HttpSinkRequestEntry> elementConverter = getElementConverter();
+        if (elementConverter instanceof SchemaLifecycleAwareElementConverter) {
+            ((SchemaLifecycleAwareElementConverter<?, ?>) elementConverter).open(context);
+        }
+        return elementConverter;
+    }
+
     @Override
     public StatefulSinkWriter<InputT, BufferedRequestState<HttpSinkRequestEntry>> restoreWriter(
                 InitContext context,
                 Collection<BufferedRequestState<HttpSinkRequestEntry>> recoveredState)
             throws IOException {
 
+        initElementConverterOfSchema(context);
+
         return new HttpSinkWriter<>(
             getElementConverter(),
             context,