You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by tzulitai <gi...@git.apache.org> on 2016/10/22 11:03:09 UTC
[GitHub] flink pull request #2681: [FLINK-4155] [kafka] Move partition list fetching ...
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 <tz...@apache.org>
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2681: [FLINK-4155] [kafka] Move partition list fetching to open...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on the issue:
https://github.com/apache/flink/pull/2681
@StephanEwen I think it makes sense to convert the table sink / source tests to unit tests that simply check they are using the Kafka consumer / producers correctly. Will include as part of this PR.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2681: [FLINK-4155] [kafka] Move partition list fetching to open...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the issue:
https://github.com/apache/flink/pull/2681
Did you refactor the table tests? If not, we can also merge this PR as is and file a follow up JIRA.
One other thing I thought about while looking over the PR: We need the partition list only for the partitioner.
So we actually need to do the getPartitions() call only if a partitioner has been set.
The PR is +1 to merge. Its up to you if you are doing the getPartition() call conditional with the PR, or whether we do it in a separate PR.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2681: [FLINK-4155] [kafka] Move partition list fetching to open...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on the issue:
https://github.com/apache/flink/pull/2681
Follow-up JIRA for conditional partition fetching:
https://issues.apache.org/jira/browse/FLINK-5128
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2681: [FLINK-4155] [kafka] Move partition list fetching to open...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the issue:
https://github.com/apache/flink/pull/2681
Looks good to me. I think it would be good if @rmetzger also had a look.
Concerning the Table sink tests and the mocking of the producer: Do these tests need a Kafka cluster running at all? They are just a think layer on top of the producer, which should be covered with Unit tests and not need a Kafka cluster.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2681: [FLINK-4155] [kafka] Move partition list fetching to open...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the issue:
https://github.com/apache/flink/pull/2681
Cool, thx.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2681: [FLINK-4155] [kafka] Move partition list fetching to open...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on the issue:
https://github.com/apache/flink/pull/2681
Actually, IIRC, @twalthr already converted the table tests to unit tests in https://issues.apache.org/jira/browse/FLINK-4745. So, right now the table tests are already unit tests without running a Kafka cluster.
The only change regarding the table tests in this PR is that on the producer side (table sink), the producer had to be mocked because of the previous partition fetching in the constructor; this PR merely changes the mocking to directly instantiating.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2681: [FLINK-4155] [kafka] Move partition list fetching to open...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on the issue:
https://github.com/apache/flink/pull/2681
Thanks for the review @rmetzger.
No, actually I haven't refactored the table tests yet. I'll file up the follow-up JIRAs mentioned in the comments, and merge the PR as is.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2681: [FLINK-4155] [kafka] Move partition list fetching ...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/2681
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---