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 2017/06/27 08:18:55 UTC
[GitHub] flink pull request #4190: [FLINK-7011] [kafka] Harden Kafka testStartFromKaf...
GitHub user tzulitai opened a pull request:
https://github.com/apache/flink/pull/4190
[FLINK-7011] [kafka] Harden Kafka testStartFromKafkaCommitOffsets ITCases
Hardens `testStartFromKafkaCommitOffsets` in Kafka ITCases.
**Description of what the test does:**
The case verifies that whatever offset was committed to Kafka, Flink reads it correctly and can use that as the correct starting point for exactly-once. It is done in an end-to-end manner, verifying that the commit logic and read offset logic is coherent.
**Problem:**
The previous implementation was too strict. It tries 3 times to fetch some committed offsets. If none is fetched, the test fails. This PR changes it so that we retry infinitely until some offsets are committed. In the case that no offsets are ever committed due to incorrect offset commit logic, the test timeouts can guard against that.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/tzulitai/flink FLINK-7011
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/4190.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 #4190
----
commit 264172d3b788ff73aade75116e8342c132df24c0
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
Date: 2017-06-27T07:53:06Z
[FLINK-7011] [kafka] Harden Kafka testStartFromKafkaCommitOffsets ITCases
----
---
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 #4190: [FLINK-7011] [kafka] Harden Kafka testStartFromKaf...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/4190#discussion_r124961658
--- Diff: flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java ---
@@ -279,49 +279,43 @@ public void runStartFromKafkaCommitOffsets() throws Exception {
final String topicName = writeSequence("testStartFromKafkaCommitOffsetsTopic", recordsInEachPartition, parallelism, 1);
- KafkaTestEnvironment.KafkaOffsetHandler kafkaOffsetHandler = kafkaServer.createOffsetHandler();
+ // read some records so that some offsets are committed to Kafka
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.getConfig().disableSysoutLogging();
+ env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
+ env.setParallelism(parallelism);
+ env.enableCheckpointing(20); // fast checkpoints to make sure we commit some offsets
- Long o1;
- Long o2;
- Long o3;
- int attempt = 0;
- // make sure that o1, o2, o3 are not all null before proceeding
- do {
- attempt++;
- LOG.info("Attempt " + attempt + " to read records and commit some offsets to Kafka");
-
- final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.getConfig().disableSysoutLogging();
- env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
- env.setParallelism(parallelism);
- env.enableCheckpointing(20); // fast checkpoints to make sure we commit some offsets
-
- env
- .addSource(kafkaServer.getConsumer(topicName, new SimpleStringSchema(), standardProps))
- .map(new ThrottledMapper<String>(consumePause))
- .map(new MapFunction<String, Object>() {
- int count = 0;
- @Override
- public Object map(String value) throws Exception {
- count++;
- if (count == recordsToConsume) {
- throw new SuccessException();
- }
- return null;
+ env
+ .addSource(kafkaServer.getConsumer(topicName, new SimpleStringSchema(), standardProps))
+ .map(new ThrottledMapper<String>(consumePause))
+ .map(new MapFunction<String, Object>() {
+ int count = 0;
+ @Override
+ public Object map(String value) throws Exception {
+ count++;
+ if (count == recordsToConsume) {
+ throw new SuccessException();
}
- })
- .addSink(new DiscardingSink<>());
+ return null;
+ }
+ })
+ .addSink(new DiscardingSink<>());
- tryExecute(env, "Read some records to commit offsets to Kafka");
+ tryExecute(env, "Read some records to commit offsets to Kafka");
+ // make sure that we indeed have some offsets committed to Kafka
+ Long o1 = null;
+ Long o2 = null;
+ Long o3 = null;
+ KafkaTestEnvironment.KafkaOffsetHandler kafkaOffsetHandler = kafkaServer.createOffsetHandler();
+ while (o1 == null && o2 == null && o3 == null) {
+ Thread.sleep(100);
o1 = kafkaOffsetHandler.getCommittedOffset(topicName, 0);
o2 = kafkaOffsetHandler.getCommittedOffset(topicName, 1);
o3 = kafkaOffsetHandler.getCommittedOffset(topicName, 2);
- } while (o1 == null && o2 == null && o3 == null && attempt < 3);
--- End diff --
I understand the argument. Perhaps it is also a fact that this test is covering too much into one single test, hence the awkwardness in making it stable.
I think it is sufficient to have 2 separate tests that replace this:
(a) test that committed Kafka offsets are correct (there is already a ITCase for this)
(b) test that committed offsets are correctly picked up and used correctly (there is actually also a test for this already).
Hence, I would conclude that perhaps this test can be removed.
---
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 #4190: [FLINK-7011] [kafka] Harden Kafka testStartFromKaf...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/4190#discussion_r124527723
--- Diff: flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java ---
@@ -279,49 +279,43 @@ public void runStartFromKafkaCommitOffsets() throws Exception {
final String topicName = writeSequence("testStartFromKafkaCommitOffsetsTopic", recordsInEachPartition, parallelism, 1);
- KafkaTestEnvironment.KafkaOffsetHandler kafkaOffsetHandler = kafkaServer.createOffsetHandler();
+ // read some records so that some offsets are committed to Kafka
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.getConfig().disableSysoutLogging();
+ env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
+ env.setParallelism(parallelism);
+ env.enableCheckpointing(20); // fast checkpoints to make sure we commit some offsets
- Long o1;
- Long o2;
- Long o3;
- int attempt = 0;
- // make sure that o1, o2, o3 are not all null before proceeding
- do {
- attempt++;
- LOG.info("Attempt " + attempt + " to read records and commit some offsets to Kafka");
-
- final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.getConfig().disableSysoutLogging();
- env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
- env.setParallelism(parallelism);
- env.enableCheckpointing(20); // fast checkpoints to make sure we commit some offsets
-
- env
- .addSource(kafkaServer.getConsumer(topicName, new SimpleStringSchema(), standardProps))
- .map(new ThrottledMapper<String>(consumePause))
- .map(new MapFunction<String, Object>() {
- int count = 0;
- @Override
- public Object map(String value) throws Exception {
- count++;
- if (count == recordsToConsume) {
- throw new SuccessException();
- }
- return null;
+ env
+ .addSource(kafkaServer.getConsumer(topicName, new SimpleStringSchema(), standardProps))
+ .map(new ThrottledMapper<String>(consumePause))
+ .map(new MapFunction<String, Object>() {
+ int count = 0;
+ @Override
+ public Object map(String value) throws Exception {
+ count++;
+ if (count == recordsToConsume) {
+ throw new SuccessException();
}
- })
- .addSink(new DiscardingSink<>());
+ return null;
+ }
+ })
+ .addSink(new DiscardingSink<>());
- tryExecute(env, "Read some records to commit offsets to Kafka");
+ tryExecute(env, "Read some records to commit offsets to Kafka");
+ // make sure that we indeed have some offsets committed to Kafka
+ Long o1 = null;
+ Long o2 = null;
+ Long o3 = null;
+ KafkaTestEnvironment.KafkaOffsetHandler kafkaOffsetHandler = kafkaServer.createOffsetHandler();
+ while (o1 == null && o2 == null && o3 == null) {
+ Thread.sleep(100);
o1 = kafkaOffsetHandler.getCommittedOffset(topicName, 0);
o2 = kafkaOffsetHandler.getCommittedOffset(topicName, 1);
o3 = kafkaOffsetHandler.getCommittedOffset(topicName, 2);
- } while (o1 == null && o2 == null && o3 == null && attempt < 3);
--- End diff --
I would still add a hard-limit for attempts (20?); there's no benefit in a test that succeeds _at some point_, if it doesn't finish relatively quickly the entire build will time out anyway.
Also, when running tests locally i suppose there isn't even a time-limit that would kill the test...
---
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 #4190: [FLINK-7011] [kafka] Harden Kafka testStartFromKaf...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/4190
---
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 #4190: [FLINK-7011] [kafka] Harden Kafka testStartFromKafkaCommi...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on the issue:
https://github.com/apache/flink/pull/4190
@zentol the time limit is set on each individual version-specific `KafkaConsumerXXITCase`
---
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.
---