You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by GitBox <gi...@apache.org> on 2020/03/11 12:25:53 UTC
[GitHub] [beam] JozoVilcek opened a new pull request #11099: [BEAM-9420]
Configurable timeout for blocking kafka API call(s)
JozoVilcek opened a new pull request #11099: [BEAM-9420] Configurable timeout for blocking kafka API call(s)
URL: https://github.com/apache/beam/pull/11099
[BEAM-9420] Configurable timeout for blocking kafka API call(s)
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
Lang | SDK | Apex | Dataflow | Flink | Gearpump | Samza | Spark
--- | --- | --- | --- | --- | --- | --- | ---
Go | [](https://builds.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/)
Java | [](https://builds.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Apex/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Gearpump/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/)
Python | [](https://builds.apache.org/job/beam_PostCommit_Python2/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python35/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/) | --- | [](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Python2_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python35_VR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/)
XLang | --- | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/)
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website
--- | --- | --- | --- | ---
Non-portable | [](https://builds.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/)
Portable | --- | [](https://builds.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/) | --- | ---
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] iemejia commented on issue #11099: [BEAM-9420] Configurable
timeout for blocking kafka API call(s)
Posted by GitBox <gi...@apache.org>.
iemejia commented on issue #11099: [BEAM-9420] Configurable timeout for blocking kafka API call(s)
URL: https://github.com/apache/beam/pull/11099#issuecomment-601928283
@aromanenko-dev maybe since he has been maintaing KafkaIo for a while.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] JozoVilcek removed a comment on issue #11099: [BEAM-9420]
Configurable timeout for blocking kafka API call(s)
Posted by GitBox <gi...@apache.org>.
JozoVilcek removed a comment on issue #11099: [BEAM-9420] Configurable timeout for blocking kafka API call(s)
URL: https://github.com/apache/beam/pull/11099#issuecomment-597719903
R: @iemejia
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] JozoVilcek commented on issue #11099: [BEAM-9420]
Configurable timeout for blocking kafka API call(s)
Posted by GitBox <gi...@apache.org>.
JozoVilcek commented on issue #11099: [BEAM-9420] Configurable timeout for blocking kafka API call(s)
URL: https://github.com/apache/beam/pull/11099#issuecomment-597719903
R: @iemejia
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] JozoVilcek commented on issue #11099: [BEAM-9420]
Configurable timeout for blocking kafka API call(s)
Posted by GitBox <gi...@apache.org>.
JozoVilcek commented on issue #11099: [BEAM-9420] Configurable timeout for blocking kafka API call(s)
URL: https://github.com/apache/beam/pull/11099#issuecomment-600113601
@kennknowles is there anything else you would like to improve in this PR?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] aromanenko-dev commented on a change in pull request #11099:
[BEAM-9420] Configurable timeout for blocking kafka API call(s)
Posted by GitBox <gi...@apache.org>.
aromanenko-dev commented on a change in pull request #11099: [BEAM-9420] Configurable timeout for blocking kafka API call(s)
URL: https://github.com/apache/beam/pull/11099#discussion_r397111316
##########
File path: sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
##########
@@ -827,6 +832,19 @@ public void setValueDeserializer(String valueDeserializer) {
return toBuilder().setOffsetConsumerConfig(offsetConsumerConfig).build();
}
+ /**
+ * Sets a default timeout to use for API calls to Kafka which are possibly blocking. {@link
+ * KafkaIO.Read} will not wait indefinitely for such calls but cancels them and fail after given
+ * timeout.
+ *
+ * <p>Higher versions of kafka client library supports such API timeout, which is passed in
+ * consumer config as `default.api.timeout.ms`. If timeout is not explicitly set here, then
+ * {@link KafkaIO.Read} will look for `default.api.timeout.ms` parameter in consumer config.
+ */
+ public Read<K, V> withDefaultKafkaApiTimeout(Duration timeout) {
Review comment:
Instead of exposing it into user API as a separate call, can we just leverage `default.api.timeout.ms` config option if it's present? If it was not set by default (old version of Kafka client, for example) then we can ask user to set explicitly with `withConsumerConfigUpdates()`? And finally, it was not set in any case, just use default value as it was before.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] aromanenko-dev commented on issue #11099: [BEAM-9420]
Configurable timeout for blocking kafka API call(s)
Posted by GitBox <gi...@apache.org>.
aromanenko-dev commented on issue #11099: [BEAM-9420] Configurable timeout for blocking kafka API call(s)
URL: https://github.com/apache/beam/pull/11099#issuecomment-603786407
Run Java PreCommit
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] aromanenko-dev merged pull request #11099: [BEAM-9420]
Configurable timeout for blocking kafka API call(s)
Posted by GitBox <gi...@apache.org>.
aromanenko-dev merged pull request #11099: [BEAM-9420] Configurable timeout for blocking kafka API call(s)
URL: https://github.com/apache/beam/pull/11099
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] aromanenko-dev commented on a change in pull request #11099:
[BEAM-9420] Configurable timeout for blocking kafka API call(s)
Posted by GitBox <gi...@apache.org>.
aromanenko-dev commented on a change in pull request #11099: [BEAM-9420] Configurable timeout for blocking kafka API call(s)
URL: https://github.com/apache/beam/pull/11099#discussion_r397121739
##########
File path: sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
##########
@@ -827,6 +832,19 @@ public void setValueDeserializer(String valueDeserializer) {
return toBuilder().setOffsetConsumerConfig(offsetConsumerConfig).build();
}
+ /**
+ * Sets a default timeout to use for API calls to Kafka which are possibly blocking. {@link
+ * KafkaIO.Read} will not wait indefinitely for such calls but cancels them and fail after given
+ * timeout.
+ *
+ * <p>Higher versions of kafka client library supports such API timeout, which is passed in
+ * consumer config as `default.api.timeout.ms`. If timeout is not explicitly set here, then
+ * {@link KafkaIO.Read} will look for `default.api.timeout.ms` parameter in consumer config.
+ */
+ public Read<K, V> withDefaultKafkaApiTimeout(Duration timeout) {
Review comment:
Thanks! Please, add a comment about that into KafkaIO class javadoc to make it clear for users.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] JozoVilcek commented on issue #11099: [BEAM-9420]
Configurable timeout for blocking kafka API call(s)
Posted by GitBox <gi...@apache.org>.
JozoVilcek commented on issue #11099: [BEAM-9420] Configurable timeout for blocking kafka API call(s)
URL: https://github.com/apache/beam/pull/11099#issuecomment-597722074
I do not see where it could help. This timeout is to just fence kafka API with some deadline. Internally, consumer will do it's own retry logic / loops of resolving the operation - e.g. seeks to initial offsets
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] JozoVilcek commented on a change in pull request #11099:
[BEAM-9420] Configurable timeout for blocking kafka API call(s)
Posted by GitBox <gi...@apache.org>.
JozoVilcek commented on a change in pull request #11099: [BEAM-9420] Configurable timeout for blocking kafka API call(s)
URL: https://github.com/apache/beam/pull/11099#discussion_r397119556
##########
File path: sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
##########
@@ -827,6 +832,19 @@ public void setValueDeserializer(String valueDeserializer) {
return toBuilder().setOffsetConsumerConfig(offsetConsumerConfig).build();
}
+ /**
+ * Sets a default timeout to use for API calls to Kafka which are possibly blocking. {@link
+ * KafkaIO.Read} will not wait indefinitely for such calls but cancels them and fail after given
+ * timeout.
+ *
+ * <p>Higher versions of kafka client library supports such API timeout, which is passed in
+ * consumer config as `default.api.timeout.ms`. If timeout is not explicitly set here, then
+ * {@link KafkaIO.Read} will look for `default.api.timeout.ms` parameter in consumer config.
+ */
+ public Read<K, V> withDefaultKafkaApiTimeout(Duration timeout) {
Review comment:
Yes, I did think of that. I was not sure if to choose that path - less transparent but easier to maintain. I suppose this can be kept at the level of "advanced configuration" which needs code familiarity. I will make these changes and push an update
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services