Skip to content

KAFKA-3311: Prepare internal source topics before calling partition grouper#990

Closed
guozhangwang wants to merge 3 commits into
apache:trunkfrom
guozhangwang:K3311
Closed

KAFKA-3311: Prepare internal source topics before calling partition grouper#990
guozhangwang wants to merge 3 commits into
apache:trunkfrom
guozhangwang:K3311

Conversation

@guozhangwang

@guozhangwang guozhangwang commented Mar 1, 2016

Copy link
Copy Markdown
Contributor

No description provided.

@guozhangwang

guozhangwang commented Mar 1, 2016

Copy link
Copy Markdown
Contributor Author

Ping @ymatsuda for reviews:

  1. For all internal source topics, pre-validate the topic with the number of partitions to:
    a. number of partitions to the co-partition topics if there is any.
    b. maximum number of partitions to the depending sub-topologies (i.e. whose sink nodes include this internal source topic).

  2. After calling partition grouper, re-validate the changelog and internal source topics to the maximum number of tasks.partitionId + 1.

  3. move the co-partition checking logic to StreamsPartitionAssignor.

@ymatsuda

ymatsuda commented Mar 1, 2016

Copy link
Copy Markdown
Contributor

What happens if auto.create.topics.enable=false?

@guozhangwang

guozhangwang commented Mar 2, 2016

Copy link
Copy Markdown
Contributor Author

@ymatsuda Right now we have two rounds for create topics:

  1. before calling partition grouper, create topics for internal source topics according to the logic above;

  2. after calling partition grouper, create / update topics for changelog and internal source topics.

So we do not need to rely on broker side auto.create.topics.enable config.

ymatsuda
ymatsuda reviewed Mar 2, 2016
View reviewed changes
Comment thread streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignor.java Outdated
int numPartitions = -1;

for (String topic : copartitionGroup) {
List<PartitionInfo> infos = metadata.partitionsForTopic(topic);

@ymatsuda ymatsuda Mar 2, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

It seems you can move this line after line422.

@guozhangwang guozhangwang Mar 2, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Ack, will fix.

@guozhangwang

guozhangwang commented Mar 2, 2016

Copy link
Copy Markdown
Contributor Author

@ymatsuda addressed comments.

@guozhangwang guozhangwang mentioned this pull request Mar 2, 2016
Merged
@ymatsuda

ymatsuda commented Mar 2, 2016

Copy link
Copy Markdown
Contributor

LGTM

@junrao

junrao commented Mar 2, 2016

Copy link
Copy Markdown
Contributor

Thanks for the patch. LGTM

@asfgit asfgit closed this in 2a58ba9 Mar 2, 2016
guozhangwang added a commit to guozhangwang/kafka that referenced this pull request Mar 3, 2016
@guozhangwang
KAFKA-3311; Prepare internal source topics before calling partition g…
fea9995 
…rouper

Author: Guozhang Wang <wangguoz@gmail.com>

Reviewers: Yasuhiro Matsuda <yasuhiro.matsuda@gmail.com>, Jun Rao <junrao@gmail.com>

Closes apache#990 from guozhangwang/K3311
@guozhangwang guozhangwang deleted the K3311 branch October 7, 2016 21:46
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

No reviews

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

3 participants

@guozhangwang @ymatsuda @junrao