GitHub user tzulitai opened a pull request:
https://github.com/apache/flink/pull/2681
[FLINK-4155] [kafka] Move partition list fetching to open() for Kafka
Producers
The fetched partition list from Kafka in `open()` is sorted by partition id
so that subtasks will have the same list across failures. To compensate the
original
use of a `KafkaProducer` instantiation in the constructor to eagerly
ensure that required producer configs are provided, we check that at least
the bootstrap servers are set.
This change also includes:
1. Refactoring of `AtLeastOnceProducerTest` for a more complete suite of
tests on `FlinkKafkaProducerBase`.
2. Directly instantiate a `FlinkKafkaProducerBase` in the Kafka table sink
tests, instead of mocking (had to be mocked before because of the
`KafkaProducer` in the constructor).
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/tzulitai/flink FLINK-4155
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/2681.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #2681
----
commit c8f152037f2f6c31be2b24613ddad8dd1881e631
Author: Tzu-Li (Gordon) Tai <[email protected]>
Date: 2016-10-21T06:23:58Z
[FLINK-4155] [kafka] Move partition list fetching to open() for Kafka
producers
The fetched partition list from Kafka in open() is sorted by partition id
so that subtasks will have the same list across failures. To compensate the
original
use of the KafkaProducer instantiation in the constructor to eagerly
ensure that required producer configs are provided, we check that at least
the bootstrap servers are set.
This change also includes refactoring of AtLeastOnceProducerTest for a more
complete suite of tests on FlinkKafkaProducerBase.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
