You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2021/02/12 02:38:54 UTC
[GitHub] [beam] rionmonster opened a new pull request #13975: BEAM-11806 - Explicit Partition Support for KafkaIO.WriteRecords
rionmonster opened a new pull request #13975:
URL: https://github.com/apache/beam/pull/13975
At present, the `WriteRecords` transform for KafkaIO does not recognize the `partition` property defined on `ProducerRecord` instances consumed by the transform:
```
producer.send(
// The null property in the following constructor represents partition
new ProducerRecord<>(
topicName, null, timestampMillis, record.key(), record.value(), record.headers()),
new SendCallback());
```
Because of this limitation, in a scenario where a user may desire an explicitly defined partitioning strategy as opposed to round-robin, they would have to create their own custom `DoFn` that defines a KafkaProducer (preferably within a @StartBundle) similar to the following approach (in Kotlin):
```
private class ExampleProducerDoFn(...): DoFn<...>() {
private lateinit var producer: KafkaProducer<...>
@StartBundle
fun startBundle(context: StartBundleContext) {
val options = context.pipelineOptions.`as`(YourPipelineOptions::class.java)
producer = getKafkaProducer(options)
}
@ProcessElement
fun processElement(context: ProcessContext){
// Omitted for brevity
// Produce the record to a specific topic at a specific partition
producer.send(ProducerRecord(
"your_topic_here",
your_partition_here,
context.element().kv.key,
context.element().kv.value
))
}
```
The _initial_ pull request that I threw in here simply replaces the existing `null` with the `record.partition()` (i.e. the record that was explicitly defined initially, but it may require some other changes which I'd need someone more familiar with the KafkaIO source to chime in on.
Additionally, I created [a corresponding JIRA ticket](https://issues.apache.org/jira/browse/BEAM-11806), which any feedback would be appreciated on as well.
------------------------
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 | Dataflow | Flink | Samza | Spark | Twister2
--- | --- | --- | --- | --- | --- | ---
Go | [](https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/) | ---
Java | [](https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://ci-beam
.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/)
Python | [](https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/) | ---
XLang | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/) | ---
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website | Whitespace | Typescript
--- | --- | --- | --- | --- | --- | ---
Non-portable | [](https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/lastCompletedBuild/) <br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/)
Portable | --- | [](https://ci-beam.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.
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
----------------------------------------------------------------
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
[GitHub] [beam] rionmonster commented on pull request #13975: [BEAM-11806] Explicit Partition Support for KafkaIO.WriteRecords
Posted by GitBox <gi...@apache.org>.
rionmonster commented on pull request #13975:
URL: https://github.com/apache/beam/pull/13975#issuecomment-778402659
@aromanenko-dev
I've made the changes that you had mentioned (added a unit test that required a few additional constructors to support the addition of partitions in the `KV2ProducerRecord` test class). I ran through the spotless apply script as well as a check against the KafkaIO SDK after the changes were made and it was successful.
Let me know if there's anything else that we need to do on this front or if I need to update the JIRA to reflect a specific version, etc.
----------------------------------------------------------------
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
[GitHub] [beam] aromanenko-dev commented on pull request #13975: [BEAM-11806] Explicit Partition Support for KafkaIO.WriteRecords
Posted by GitBox <gi...@apache.org>.
aromanenko-dev commented on pull request #13975:
URL: https://github.com/apache/beam/pull/13975#issuecomment-778275873
Rion, please add your test into `org.apache.beam.sdk.io.kafka.KafkaIOTest`, we used to use to KafkaIO unit testing in general. I believe it already contains some tests for Write with `ProducerRecord`, take a look on `testRecordsSink()`, for example. I think it can be just adjusted to check that right partition was properly set in `ProducerRecord`s.
----------------------------------------------------------------
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
[GitHub] [beam] aromanenko-dev commented on pull request #13975: [BEAM-11806] Explicit Partition Support for KafkaIO.WriteRecords
Posted by GitBox <gi...@apache.org>.
aromanenko-dev commented on pull request #13975:
URL: https://github.com/apache/beam/pull/13975#issuecomment-779147363
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
[GitHub] [beam] aromanenko-dev commented on pull request #13975: [BEAM-11806] Explicit Partition Support for KafkaIO.WriteRecords
Posted by GitBox <gi...@apache.org>.
aromanenko-dev commented on pull request #13975:
URL: https://github.com/apache/beam/pull/13975#issuecomment-778207990
Run Java PostCommit
----------------------------------------------------------------
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
[GitHub] [beam] aromanenko-dev commented on pull request #13975: [BEAM-11806] Explicit Partition Support for KafkaIO.WriteRecords
Posted by GitBox <gi...@apache.org>.
aromanenko-dev commented on pull request #13975:
URL: https://github.com/apache/beam/pull/13975#issuecomment-779186048
LGTM, merging. I'll take care of Jira.
Thank you for contribution!
----------------------------------------------------------------
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
[GitHub] [beam] aromanenko-dev merged pull request #13975: [BEAM-11806] Explicit Partition Support for KafkaIO.WriteRecords
Posted by GitBox <gi...@apache.org>.
aromanenko-dev merged pull request #13975:
URL: https://github.com/apache/beam/pull/13975
----------------------------------------------------------------
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
[GitHub] [beam] rionmonster commented on pull request #13975: [BEAM-11806] Explicit Partition Support for KafkaIO.WriteRecords
Posted by GitBox <gi...@apache.org>.
rionmonster commented on pull request #13975:
URL: https://github.com/apache/beam/pull/13975#issuecomment-778252803
Thanks Alexey,
I've updated the original commit message to include the JIRA issue as recommended. Do you know of the more appropriate place within the repository to add the unit test? I see quite a bit of Kafka-related tests across the place, but wasn't sure where something like this would best fit.
Additionally, I do feel that the recommended refactoring would be most appropriate in this case (i.e. sending the `ProducerRecord` instance itself instead of creating a new instance). IMO, that change makes more sense, but I'll leave it up to you if that's the referred fix here.
----------------------------------------------------------------
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
[GitHub] [beam] aromanenko-dev edited a comment on pull request #13975: [BEAM-11806] Explicit Partition Support for KafkaIO.WriteRecords
Posted by GitBox <gi...@apache.org>.
aromanenko-dev edited a comment on pull request #13975:
URL: https://github.com/apache/beam/pull/13975#issuecomment-778275873
Rion, please add your test into `org.apache.beam.sdk.io.kafka.KafkaIOTest`, we used to use for KafkaIO unit testing in general. I believe it already contains some tests for `ProducerRecord` sink - please, take a look on `testRecordsSink()`, for example. I think it can be just adjusted to check that right partition was properly set in `ProducerRecord`s.
----------------------------------------------------------------
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