[
https://issues.apache.org/jira/browse/FLINK-4155?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15597660#comment-15597660
]
ASF GitHub Bot commented on FLINK-4155:
---------------------------------------
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.
----
> Get Kafka producer partition info in open method instead of constructor
> -----------------------------------------------------------------------
>
> Key: FLINK-4155
> URL: https://issues.apache.org/jira/browse/FLINK-4155
> Project: Flink
> Issue Type: Bug
> Components: Kafka Connector
> Affects Versions: 1.1.0, 1.0.3
> Reporter: Gyula Fora
> Assignee: Tzu-Li (Gordon) Tai
> Fix For: 1.2.0
>
>
> Currently the Flink Kafka producer does not really do any error handling if
> something is wrong with the partition metadata as it is serialized with the
> user function.
> This means that in some cases the job can go into an error loop when using
> the checkpoints. Getting the partition info in the open method would solve
> this problem (like restarting from a savepoint which re-runs the constructor).
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
