streamnative#386: Check serializability of crypto key reader and encryption keys.

Author: objecttrouve

pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/FlinkPulsarSink.java

@@ -35,7 +35,9 @@
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 
+import static org.apache.flink.util.InstantiationUtil.isSerializable;
 import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
 
 /**
  * Write data to Flink.
@@ -136,6 +138,8 @@ public FlinkPulsarSink<T>  build(){
             if ((cryptoKeyReader == null) != (encryptionKeys.isEmpty())){
                 throw new IllegalStateException("Set crypto key reader and encryption keys in conjunction.");
             }
+            checkState(isSerializable(cryptoKeyReader));
+            checkState(isSerializable(encryptionKeys));
             return new FlinkPulsarSink<>(this);
         }
 

pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/FlinkPulsarSource.java

@@ -88,8 +88,11 @@
 import static org.apache.flink.streaming.connectors.pulsar.internal.metrics.PulsarSourceMetrics.COMMITS_FAILED_METRICS_COUNTER;
 import static org.apache.flink.streaming.connectors.pulsar.internal.metrics.PulsarSourceMetrics.COMMITS_SUCCEEDED_METRICS_COUNTER;
 import static org.apache.flink.streaming.connectors.pulsar.internal.metrics.PulsarSourceMetrics.PULSAR_SOURCE_METRICS_GROUP;
+import static org.apache.flink.util.InstantiationUtil.isSerializable;
 import static org.apache.flink.util.Preconditions.checkArgument;
 import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
 
 /**
  * Pulsar data source.
@@ -168,6 +171,7 @@ public FlinkPulsarSource<T> build(){
             if (clientConf == null){
                 throw new IllegalStateException("Client conf mustn't be null. Either provide a client conf or a service URL plus properties.");
             }
+            checkState(isSerializable(cryptoKeyReader));
             return new FlinkPulsarSource<>(this);
         }