Check if producer is not null before closing
@@ -65,9 +65,11 @@ public abstract class KafkaSink<K, V> extends SimpleSink<byte[]> {
@Override
public void close() throws IOException {
+ if (producer != null) {
producer.close();
LOG.info("Kafka sink stopped.");
}
+ }
@Override
public void open(Map<String, Object> config) throws Exception {
