### Question:
FlinkKinesisProducer extends RichSinkFunction<OUT> implements CheckpointedFunction { public void setCustomPartitioner(KinesisPartitioner<OUT> partitioner) { checkNotNull(partitioner, "partitioner cannot be null"); checkArgument( InstantiationUtil.isSerializable(partitioner), "The provided custom partitioner is not serializable: " + partitioner.getClass().getName() + ". " + "Please check that it does not contain references to non-serializable instances."); this.customPartitioner = partitioner; } FlinkKinesisProducer(final SerializationSchema<OUT> schema, Properties configProps);  FlinkKinesisProducer(KinesisSerializationSchema<OUT> schema, Properties configProps); void setFailOnError(boolean failOnError); void setDefaultStream(String defaultStream); void setDefaultPartition(String defaultPartition); void setCustomPartitioner(KinesisPartitioner<OUT> partitioner); @Override void open(Configuration parameters); @Override void invoke(OUT value, Context context); @Override void close(); @Override void initializeState(FunctionInitializationContext context); @Override void snapshotState(FunctionSnapshotContext context);  }### Answer:
@Test public void testConfigureWithNonSerializableCustomPartitionerFails() { exception.expect(IllegalArgumentException.class); exception.expectMessage("The provided custom partitioner is not serializable"); new FlinkKinesisProducer<>(new SimpleStringSchema(), TestUtils.getStandardProperties()) .setCustomPartitioner(new NonSerializableCustomPartitioner()); }
@Test public void testConfigureWithSerializableCustomPartitioner() { new FlinkKinesisProducer<>(new SimpleStringSchema(), TestUtils.getStandardProperties()) .setCustomPartitioner(new SerializableCustomPartitioner()); }