[FLINK-25726][streaming] Check committer existence in SinkV1Adapter t…

…o determine if a committer is needed

Before this commit the serializer presence was the indicator.
Unfortunately, the serializer is also required in case of using a global
committer. The topology having a single writer and a global committer in
Sink V1 would have failed.

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SinkV1Adapter.java

@@ -88,8 +88,12 @@ public Sink<InputT> asSpecializedSink() {
         if (sink.getGlobalCommittableSerializer().isPresent()) {
             globalCommitter = true;
         }
-        if (sink.getCommittableSerializer().isPresent()) {
-            committer = true;
+        try {
+            if (sink.createCommitter().isPresent()) {
+                committer = true;
+            }
+        } catch (IOException e) {
+            throw new IllegalStateException("Failed to instantiate committer.", e);
         }
 
         if (globalCommitter && committer && stateful) {

flink-streaming-java/src/test/java/org/apache/flink/streaming/api/transformations/SinkV1AdapterTest.java

@@ -128,6 +128,22 @@ private static class CommittingSinkV1 extends DefaultSinkV1 {
         public Optional<SimpleVersionedSerializer<Integer>> getCommittableSerializer() {
             return Optional.of(new NoOpSerializer());
         }
+
+        @Override
+        public Optional<Committer<Integer>> createCommitter() throws IOException {
+            return Optional.of(
+                    new Committer<Integer>() {
+                        @Override
+                        public List<Integer> commit(List<Integer> committables) {
+                            throw new UnsupportedOperationException();
+                        }
+
+                        @Override
+                        public void close() {
+                            throw new UnsupportedOperationException();
+                        }
+                    });
+        }
     }
 
     private static class StatefulCommittingSinkV1 extends CommittingSinkV1 {