KAFKA-7515: Trogdor - Add Consumer Group Benchmark Specification by stanislavkozlovski · Pull Request #5810 · apache/kafka

stanislavkozlovski · 2018-10-17T09:13:25Z

https://issues.apache.org/jira/browse/KAFKA-7515

Changes

Add new consumerGroup field to ConsumeBenchSpec.
Changes the activeTopics field format in ConsumeBenchSpec.
- activeTopics is a list of strings and now supports three notations for each value
  - 'foo' - denotes a topic name 'foo'
  - single-range notation 'foo[1-2] - gets expanded to two topics - 'foo1' and 'foo2'
  - double-range notation 'foo[1-2][1-2] - gets expanded to two topics with two partitions each. topic 'foo1' with partitions 1 and 2, topic 'foo2' with partitions 1 and 2

This ConsumeBenchWorker now supports three cases:

When we want to manually assign partitions and use a random, new consumer group. (activeTopics contains at least one value with the new double-range notation (e.g foo[1-2][1-2]), consumerGroup is undefined) - consumer uses the specific partitions (and all partitions for topics who did not have in a second range) via KafkaConsumer#assign()
When we want to have dynamic partition assignment via an existing consumer group's (activeTopics does not contain any double range notations, consumerGroup is specified) - KafkaConsumer#subscribe()
When we want to manually assign partitions but track offsets via an existing consumer group (activeTopics contains at least one value with the new double-range notation, consumerGroup is specified) - KafkaConsumer#assign()

stanislavkozlovski · 2018-10-17T15:15:14Z

cc @cmccabe @apovzner

cmccabe · 2018-10-17T17:47:16Z

Thanks, @stanislavkozlovski

It seems like the only difference between the version with a group and the version without is that we call subscribe. So we don’t need a new class, right, just a new configuration option for ConsumeBench, I think.

cmccabe · 2018-10-18T18:47:30Z

-                executor.submit(new ConsumeMessages(partitions));
+
+                AbstractConsumeMessages consumeMessagesTask;
+                if (spec.consumerGroup() == null) {


We don't use null entries in JSON, because it gets too confusing. You should check against empty string here.

oh yeah, I could just set it to an empty string when it's null in the spec - way better

Using empty string for null is more confusing no?

There's a pattern for all of the Trogdor JSON code where we don't use null anywhere. The problem with null is it gets annoying to check each collection for empty vs. null, each string for empty vs. null, etc. etc.

null is also handled kind of inconsistently in Jackson. Sometimes Jackson will serialize a field that is null as "foo": null whereas sometimes it will just omit the field. (I think that "foo": null is actually not conforming JSON, by the way...) There are probably ways to configure all this, but null doesn't really provide any value 99% of the time, so it's simpler to just treat empty as null.

We should be consistent, I agree. I don't agree with the value part. It's like saying that we should use empty String to represent the absence of a value in Java. Something like Optional is better and maps to null in JSON. Anyway, a discussion for a different venue. :)

cmccabe · 2018-10-18T18:48:09Z

+
+                AbstractConsumeMessages consumeMessagesTask;
+                if (spec.consumerGroup() == null) {
+                    spec.consumerGroup(DEFAULT_CONSUMER_GROUP);


We should use a randomly generated (and hopefully unique!) consumer group here so that we don't conflict with other people running a test.

cmccabe · 2018-10-18T18:48:46Z

        this.commonClientConf = configOrEmptyMap(commonClientConf);
        this.adminClientConf = configOrEmptyMap(adminClientConf);
        this.activeTopics = activeTopics == null ? TopicsSpec.EMPTY : activeTopics.immutableCopy();
+        this.consumerGroup = consumerGroup;


Should be consumerGroup == null ? "" : consumerGroup to match the other entries. We don't use nulls in JSON

cmccabe · 2018-10-18T18:50:43Z

+        Properties consumerProperties;

-        ConsumeMessages(Collection<TopicPartition> topicPartitions) {
+        AbstractConsumeMessages(Map<String, List<TopicPartition>> topicPartitionsByTopic) {


Seems like we don't really need inheritance here. Can just have an "if" statement that checks if we have a group or not

cmccabe · 2018-10-18T18:52:33Z

+                .flatMap(List::stream).collect(Collectors.toList());
+            KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(
+                consumerProperties, new ByteArrayDeserializer(), new ByteArrayDeserializer());
+            consumer.assign(topicPartitions);


We shouldn't always use assign here. If the developer has not specified any partitions, we can use the partitions of the group itself.

cmccabe · 2018-10-18T18:56:57Z

Thanks, @stanislavkozlovski.

So there are basically three cases here:

Developer specifies some partitions, but no group ID.
Developer specifies just a group ID
developers specifies some partitions, and also a group ID.

In case 1, we want the group ID to be randomly assigned and not to conflict with other group IDs in the cluster. Otherwise we may impact concurrently running tests.

In case 2, we want the group ID to be set. The client will ask the group for a partition assignment, rather than creating one manually. This is the most common way to use groups.

In case 3, we will set the group ID, but use KafkaConsumer#assign() to manually specify our partition assignment. In this case, the only thing we're using the group for is identifying the partition offset data we save during periodic offset auto-commits.

We can assume that we're in case 1 when we have an empty group ID.

We can assume that we're in case 2 when we have a non-empty group ID, but not any partition data. After all, it doesn't make sense to consume nothing!

Otherwise we're in case 3.

stanislavkozlovski · 2018-10-18T19:51:58Z

Thanks for the review @cmccabe,

Case 1. Agree, it makes sense to improve the current behavior and assign a random group when one isn't specified

Case 2. This also makes sense to me - we should support a consumer to "bootstrap" itself onto an existing consumer group

Case 3. I see the use case of saving offsets in that way, but I'm concerned we don't have a way to create a new consumer group from this tool. How would we go on about creating a new consumer group that it subscribed to some topics?
We would have a way to attach a consumer to a group (Case 2) but seemingly lack a way to start said group - leaving us to rely on something outside Trogdor I think.

cmccabe · 2018-10-18T21:22:49Z

I'm concerned we don't have a way to create a new consumer group from this tool.

If you join a consumer group that doesn't already exist, then it is created.

How would we go on about creating a new consumer group that it subscribed to some topics?

That is a good point. Perhaps we should have some way of support a use-case 4: create a group and subscribe (not assign) some partitions.

One way of doing this would be adding a new configuration like ignoreGroupPartitionAssignment, which could default to false, Then if it were set to true, we'd use assign; false, subscribe. What do you think?

stanislavkozlovski · 2018-10-19T08:56:09Z

If you join a consumer group that doesn't already exist, then it is created.

Exactly, but if you are creating that and haven't populated the activeTopics field nothing would happen, right?

I think supporting use-case 4 is very important and as you suggested, should be the default behavior. I'm thinking of adding a useGroupPartitionAssignment which defaults to true -> calling subscribe().

Now I think this could have some impact on existing users as they would change from using assign() on a couple of partitions to using subscribe() on the whole topics themselves. So we should make sure that's not the case

Here's how I envision the configs to work:
Case 1: activeTopics specified, consumerGroupId not -> use assign() with a random consumer group id. Here, useGroupPartitionAssignment is totally ignored. This retains the old behavior
Case 2: consumerGroupId specified, activeTopics not -> ask group for partition assignment. I think that for completeness here we should enforce useGroupPartitionAssignment to be true (which is the default). Maybe throw an error if it's false?
Case 3: consumerGroupId specified, activeTopics specified, useGroupPartitionAssignment is explicitly set to false -> use assign()
Case 4: consumerGroupId specified, activeTopics specified, useGroupPartitionAssignment is true -> use subscribe() with the given topics

stanislavkozlovski · 2018-10-19T16:29:36Z

After a bit of investigation, it seems like we cannot simply call subscribe() without any topics. This makes case 2 invalid, we always need to have some topics or partitions to subscribe/assign to. I think we should discard that case

This ConsumeBenchWorker now supports three cases: 1. When we want to manually assign partitions and use a random, new consumer group. (useGroupPartitionAssignment=false, consumerGroup is undefined) - KafkaConsumer#assign() 2. When we want to have dynamic partition assignment via an existing consumer group's (useGroupPartitionAssignment=false, consumerGroup is specified) - KafkaConsumer#subscribe() 3. When we want to manually assign partitions but track offsets via an existing consumer group (useGroupPartitionAssignment=true, consumerGroup is specified) - KafkaConsumer#assign() Adds one new field to the ConsumeBenchSpec - "useGRoupPartitionAssignment"

stanislavkozlovski · 2018-10-21T11:16:15Z

@cmccabe I've updated the PR to support cases 1, 3 and 4. Let me know if I'm on the right track and I'll make sure to update the tests/docs as well

stanislavkozlovski · 2018-10-21T11:17:28Z

+                    topics, consumerGroup);
+                consumer.subscribe(topics);
+            }
+            else {


whoops, will change to be on the same line as closing bracket

cmccabe · 2018-10-24T16:45:58Z

    private final Map<String, String> commonClientConf;
-    private final TopicsSpec activeTopics;
+    private final List<String> activeTopics;
+    private Map<String, List<TopicPartition>> materializedTopics;


Should be final

cmccabe · 2018-10-24T16:46:03Z

-    private final TopicsSpec activeTopics;
+    private final List<String> activeTopics;
+    private Map<String, List<TopicPartition>> materializedTopics;
+    private boolean useGroupPartitionAssignment;


Should be final

cmccabe · 2018-10-24T16:46:10Z

+    private final List<String> activeTopics;
+    private Map<String, List<TopicPartition>> materializedTopics;
+    private boolean useGroupPartitionAssignment;
+    private String consumerGroup;


Should be final

cmccabe · 2018-10-24T16:56:11Z

+        }
+
+        private String generateConsumerGroup() {
+            return "consumer-group-" + UUID.randomUUID().toString();


perhaps "consume-bench-" + UUID... so that it's clear that Trogdor created it?

cmccabe · 2018-10-24T16:58:36Z

Thanks, @stanislavkozlovski , this looks good. I think we're getting close.

With regard to the topics / partitions specification. The current approach in the PR, if I understand correctly, would require me to specify foo[0][0-1] if I wanted partitions foo0:0, foo0:1. That seems awkward. In general I don't think that we should conflate globs with partition numbers. What I mean is that we should allow people to specify partitions by number if there are zero, one, or two globs in the string.

I think we can do this by treating : as a partition number specifier. So for example foo:0 is partition 0 of foo. foo:[0-1] is foo:0 and foo:1, foo[0-1]:[0-1] is foo0:0, foo0:1, foo1:0, foo1:1, etc.

Supporting multiple globs in strings should be pretty simple too-- we just keep calling expand on the same set of strings until the set of strings doesn't change.

stanislavkozlovski · 2018-10-25T18:35:48Z

Thanks for the review @cmccabe. I like this notation better.
One thing I didn't implement was support for multiple, non-consecutive partitions (e.g foo:1:3:5). That should be easy to add but I feel it might overcomplicate things and be unnecessary

cmccabe · 2018-10-26T17:52:39Z

+1. Thanks, @stanislavkozlovski

stanislavkozlovski · 2018-10-29T12:46:31Z

Java 11 build timed out

22:51:49 Build timed out (after 180 minutes). Marking the build as aborted.

The Java 8 build passed fine. Maybe we could merge this?

cmccabe · 2018-10-29T17:50:12Z

Build timeout in jdk11 is unrelated. Will merge. Thanks, @stanislavkozlovski

…_group_partitions_should_raise (#6015) This is the error message we're after: "You may not specify an explicit partition assignment when using multiple consumers in the same group." We apparently changed it midway through #5810 and forgot to update the test. Reviewers: Dhruvil Shah <dhruvil@confluent.io>, Ismael Juma <ismael@juma.me.uk>

…che#5810) This ConsumeBenchWorker now supports using consumer groups. The groups may be either used to store offsets, or as subscriptions.

…_group_partitions_should_raise (apache#6015) This is the error message we're after: "You may not specify an explicit partition assignment when using multiple consumers in the same group." We apparently changed it midway through apache#5810 and forgot to update the test. Reviewers: Dhruvil Shah <dhruvil@confluent.io>, Ismael Juma <ismael@juma.me.uk>

KAFKA-7515: Add consumer group functionality to Trogdor benchmark

f3099ff

cmccabe reviewed Oct 18, 2018

View reviewed changes

stanislavkozlovski commented Oct 21, 2018

View reviewed changes

stanislavkozlovski added 2 commits October 23, 2018 13:17

KAFKA-7515: Update StringExpander to handle multiple ranges

09fdd24

KAFKA-7515 Update ConsumeBenchWorker to handle multiple ranges

ed40b54

cmccabe reviewed Oct 24, 2018

View reviewed changes

Small stylistic changes

834733a

stanislavkozlovski added 2 commits October 25, 2018 21:38

Update with latest range specification

140e77f

Simplify StringExpander implementation

23cc158

cmccabe merged commit d28c534 into apache:trunk Oct 29, 2018

stanislavkozlovski mentioned this pull request Dec 8, 2018

MINOR: Fix failing test in ConsumeBenchTest:test_multiple_consumers_specifie… #6015

Merged

Conversation

stanislavkozlovski commented Oct 17, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Changes

Uh oh!

stanislavkozlovski commented Oct 17, 2018

Uh oh!

cmccabe commented Oct 17, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ijuma Oct 20, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cmccabe commented Oct 18, 2018

Uh oh!

stanislavkozlovski commented Oct 18, 2018

Uh oh!

cmccabe commented Oct 18, 2018

Uh oh!

stanislavkozlovski commented Oct 19, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

stanislavkozlovski commented Oct 19, 2018

Uh oh!

stanislavkozlovski commented Oct 21, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cmccabe commented Oct 24, 2018

Uh oh!

stanislavkozlovski commented Oct 25, 2018

Uh oh!

cmccabe commented Oct 26, 2018

Uh oh!

stanislavkozlovski commented Oct 29, 2018

Uh oh!

cmccabe commented Oct 29, 2018

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

stanislavkozlovski commented Oct 17, 2018 •

edited

Loading

ijuma Oct 20, 2018 •

edited

Loading

stanislavkozlovski commented Oct 19, 2018 •

edited

Loading

stanislavkozlovski commented Oct 21, 2018 •

edited

Loading