-
Notifications
You must be signed in to change notification settings - Fork 3.6k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Add Flink - Pulsar Batch Sink Support #2979
Add Flink - Pulsar Batch Sink Support #2979
Conversation
} | ||
|
||
@Override | ||
public void close() throws IOException { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
close the producer when the output format is closed?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@sijie Thanks for the review this.
This is the my concern as well. However, afais in local tests, close
is called multiple times and causing to be closed the Producer
before submitting all batch.
|
||
private transient Function<Throwable, MessageId> failureCallback; | ||
|
||
private static volatile Producer<byte[]> producer; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
any reason why you make it static
?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
As same with close
function, open
function is also called multiple times. To avoid multiple Producer
instance creation, it needs to be singleton at class level so it is static
.
2f8cafb
to
c3140d5
Compare
Motivation
This PR aims to bring Flink - Pulsar
Batch Sink
Support. If user works with FlinkDataSet
API and would like to write theseDataSets
to Pulsar, this sink can help.Ref: Flink Batch Sink API
Modifications
Please find the change-set as follows:
PulsarOutputFormat
to write Flink BatchDataSets
into Pulsar.FlinkPulsarBatchSinkExample
to show how to use and to be used by users.README.md
documentationjavadoc
fix