Fix: NPE in pulsar source close
@@ -142,6 +142,7 @@ public class PulsarSource<T> extends PushSource<T> implements MessageListener<T>
@Override
public void close() throws Exception {
+ if (inputConsumers != null ) {
inputConsumers.forEach(consumer -> {
try {
consumer.close();
@@ -149,6 +150,7 @@ public class PulsarSource<T> extends PushSource<T> implements MessageListener<T>
}
});
}
+ }
@SuppressWarnings("unchecked")
@VisibleForTesting
