Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[BEAM-5798] Add support to write to multiple topics with KafkaIO #7371

Closed
wants to merge 4 commits into from
Closed

[BEAM-5798] Add support to write to multiple topics with KafkaIO #7371

wants to merge 4 commits into from

Conversation

aromanenko-dev
Copy link
Contributor

Use topic name from ProducerRecord as a destination topic

Follow this checklist to help us incorporate your contribution quickly and easily:

  • 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.
  • If this contribution is large, please file an Apache Individual Contributor License Agreement.

It will help us expedite review of your Pull Request if you tag someone (e.g. @username) to look at it.

Post-Commit Tests Status (on master branch)

Lang SDK Apex Dataflow Flink Gearpump Samza Spark
Go Build Status --- --- --- --- --- ---
Java Build Status Build Status Build Status Build Status Build Status Build Status Build Status Build Status
Python Build Status --- Build Status
Build Status		 Build Status --- --- ---

@aromanenko-dev
Copy link
Contributor Author

R: @rangadi
Please, take a look

rangadi
rangadi suggested changes Jan 3, 2019
View reviewed changes
Copy link
Contributor

@rangadi rangadi left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Left a couple of comments.
So multiple topics were not supported in #7052 ?. This is follow up to it?

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaWriter.java Show resolved Hide resolved
sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java Outdated Show resolved Hide resolved
@aromanenko-dev
Copy link
Contributor Author

aromanenko-dev commented Jan 4, 2019 •
edited

@rangadi Thanks for review. Yes, multiple output topics were not supported, I decided to split these tasks. Now, topic name from ProducerRecord will have more priority than withTopic().

@aromanenko-dev
Copy link
Contributor Author

aromanenko-dev commented Jan 4, 2019 •
edited

I also updated javadoc for WriteRecords.withTopic() and added a test for custom ts in ProducerRecords that can be specified by user.

rangadi
rangadi suggested changes Jan 4, 2019
View reviewed changes
Copy link
Contributor

@rangadi rangadi left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Over all LGTM.
Left a couple of comments.

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaWriter.java Outdated
? spec.getPublishTimestampFunction()
.getTimestamp(record, ctx.timestamp())
.getMillis()
: System.currentTimeMillis());
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No need to use currentTimeMillis(). We should leave it 'null' such that Kafka client decide.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agree, fixed

sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java Outdated

completionThread.shutdown();

// Verify that appropriate messages are written to different Kafka topics
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Update comment to say it checks for timestamp.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed

@aromanenko-dev
Copy link
Contributor Author

@rangadi Thanks, I addressed your comments

@iemejia iemejia changed the title [BEAM-5798] Add support of multiple Kafka output topics [BEAM-5798] Add support for dynamic destinations when writing to Kafka Jan 8, 2019
@iemejia iemejia self-requested a review January 8, 2019 13:30
@iemejia iemejia changed the title [BEAM-5798] Add support for dynamic destinations when writing to Kafka [BEAM-5798] Add support to write to multiple topics with KafkaIO Jan 8, 2019
iemejia
iemejia approved these changes Jan 8, 2019
View reviewed changes
Copy link
Member

@iemejia iemejia left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM

iemejia added a commit that referenced this pull request Jan 8, 2019
@iemejia
Merge pull request #7371: [BEAM-5798] Add support to write to multipl…
e124ace 
…e topics with KafkaIO
@iemejia
Copy link
Member

iemejia commented Jan 8, 2019

Merged manually to update the name of the commit and squash the extra commits.

@iemejia iemejia closed this Jan 8, 2019
@iemejia
Copy link
Member

iemejia commented Jan 8, 2019

@rangadi I renamed this PR/issue to say explicitly multiple topics because it does not align exactly with the dynamic destination style APIs in BigQueryIO, TextIO and AvroIO. I think it is probably worth to add a similar pattern here, or at least document how to achieve it if not. I filled BEAM-6385. WDYT ?
Oups and in the mean time I assigned this issue to you @aromanenko-dev,

@rangadi
Copy link
Contributor

rangadi commented Jan 8, 2019

Thanks. Renaming makes sense.
Thanks for BEAM-6385. Supporting an existing pattern make sense. The current changes are still useful since wanted to give user flexibility to write Kafka producer records so that they have full control (e.g. adding headers to records).

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Reviewers

@rangadi rangadi rangadi requested changes

@iemejia iemejia iemejia approved these changes

Assignees
No one assigned
Labels
None yet
Projects
None yet
Milestone
No milestone
Development

Successfully merging this pull request may close these issues.

None yet
3 participants
@aromanenko-dev @iemejia @rangadi