[pulsar-client] Fix NPE in pulsar bolt while publishing messages
@@ -284,7 +284,7 @@ public class TypedMessageBuilderImpl<T> implements TypedMessageBuilder<T> {
public Message<T> getMessage() {
beforeSend();
- return MessageImpl.create(msgMetadata, content, schema, producer.topic);
+ return MessageImpl.create(msgMetadata, content, schema, producer != null ? producer.getTopic() : null);
}
public long getPublishTime() {
