KAFKA-6774: Improve the default group id behavior in KafkaConsumer (KIP-289) by vahidhashemian · Pull Request #5877 · apache/kafka

vahidhashemian · 2018-11-04T05:23:24Z

Improve the default group id behavior by

changing the default consumer group to null, where no offset commit or fetch, or group management operations are allowed
deprecate the use of empty ("") consumer group on the client

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

hachikuji

@vahidhashemian Thanks for the PR. I think we also might need to hook into the logic for setting the initial position of the consumer. By default, we attempt to fetch committed offsets, but if the group id is null, we should just skip this step.

hachikuji · 2018-11-07T00:14:11Z

Perhaps group management or the the offset commit api?

vahidhashemian · 2018-11-07T06:54:29Z

@hachikuji Thanks for reviewing the PR. Submitted a patch to address your feedback.

hachikuji

Left a couple more comments. Do we have an integration test that covers basic fetching using a null group id? We should probably make sure we have at least one case which uses null and one which uses the empty string.

hachikuji · 2018-11-07T17:20:01Z

I think it's fine to change the default as long as the user hasn't explicitly provided the configuration. However, if they have explicitly set enable.auto.commit=true, then it might be better to raise an exception.

hachikuji · 2018-11-07T17:33:02Z

We should probably have a fail below if this returns without errors?

Thanks! Totally missed them.

vahidhashemian

@hachikuji I tried to address your recent comments in the new commit. Thanks for the additional feedback.

vahidhashemian · 2018-11-08T06:05:11Z

Thanks! Totally missed them.

hachikuji

Thanks for the tests. Left a few minor comments.

hachikuji · 2018-11-10T08:03:12Z

Is it possible to just pass "groupId=null" as an override?

Doesn't seem so, I get a NPE:

java.lang.NullPointerException at java.base/java.util.concurrent.ConcurrentHashMap.putVal(ConcurrentHashMap.java:1021) at java.base/java.util.concurrent.ConcurrentHashMap.put(ConcurrentHashMap.java:1016) at java.base/java.util.Properties.put(Properties.java:1309) at kafka.api.PlaintextConsumerTest.testConsumingWithNullGroupId(PlaintextConsumerTest.scala:1838) ...

hachikuji · 2018-11-10T08:03:58Z

I suppose we could leave this out?

That's correct.

hachikuji · 2018-11-10T08:05:40Z

Seems more accurate to say that consumer 3 begins from an explicit seek?

hachikuji · 2018-11-10T08:10:20Z

nit: maybe better to turn this into explicit equals assertions so that we know which check failed.

Sounds good.

hachikuji · 2018-11-10T08:11:53Z

The third iteration here feels redundant. Doesn't it just read from the committed offset like consumer2?

Yes, I put it to verify the auto.offset.reset doesn't affect the behavior. I'll remove it in the new commit.

vahidhashemian

@hachikuji I updated the unit tests based on your comments. Thanks!

vahidhashemian · 2018-11-10T17:58:03Z

That's correct.

vahidhashemian · 2018-11-10T18:00:42Z

Doesn't seem so, I get a NPE:

java.lang.NullPointerException at java.base/java.util.concurrent.ConcurrentHashMap.putVal(ConcurrentHashMap.java:1021) at java.base/java.util.concurrent.ConcurrentHashMap.put(ConcurrentHashMap.java:1016) at java.base/java.util.Properties.put(Properties.java:1309) at kafka.api.PlaintextConsumerTest.testConsumingWithNullGroupId(PlaintextConsumerTest.scala:1838) ...

vahidhashemian · 2018-11-10T18:04:45Z

Yes, I put it to verify the auto.offset.reset doesn't affect the behavior. I'll remove it in the new commit.

vahidhashemian · 2018-11-10T18:05:11Z

Sounds good.

hachikuji

@vahidhashemian Thanks for the patch. Looks good overall. I had just one thought. When the groupId is null, there should be no coordinator interaction at all. We could make this explicit by not bothering to construct the ConsumerCoordinator object. That would take it out of the poll path and reduce the risk of using it accidentally. What do you think?

vahidhashemian · 2018-11-13T05:11:33Z

@hachikuji thanks for the feedback. Your suggestion makes sense. Even though it's not explicitly mentioned in the KIP, I think it wouldn't result in a different behavior that what is covered by the KIP. I submitted a patch that should cover it (it did not break any test locally), but please let me know if I missed something. Thanks!

hachikuji

Thanks for the updates, just a few additional comments.

hachikuji · 2018-11-13T18:33:48Z

Perhaps now we don't need this? We could add an assertion in the constructor that the groupId is not null.

You're right. I added an assertion in AbstractCoordinator constructor. It seemed to be a better fit there.

hachikuji · 2018-11-13T18:34:33Z

nit: not sure if there's a strong reason for any of these to use the empty group id. Could we use groupId instead?

No particular reason other than I wanted to stick to the actual group id that was used before. I updated these to use the provided groupId.

hachikuji · 2018-11-13T18:38:09Z

I was wondering if we could rephrase this to suggest what the user should do more clearly. For example, "To use the group management or offset commit APIs, you must provide a valid group.id in the consumer configuration."

Sounds good, thanks!

hachikuji · 2018-11-13T18:38:43Z

Just to complete the integration picture, could we make a call to committed and assert the exception?

I did that in the next commit.

vahidhashemian

@hachikuji thanks for the feedback/suggestions. The new commit should address them.

vahidhashemian · 2018-11-14T02:11:20Z

You're right. I added an assertion in AbstractCoordinator constructor. It seemed to be a better fit there.

vahidhashemian · 2018-11-14T02:12:02Z

Sounds good, thanks!

vahidhashemian · 2018-11-14T02:16:28Z

No particular reason other than I wanted to stick to the actual group id that was used before. I updated these to use the provided groupId.

vahidhashemian · 2018-11-14T02:28:14Z

I did that in the next commit.

hachikuji · 2018-11-14T17:26:04Z

@vahidhashemian Can you look into the build failures? I looked at one of the streams tests (KTableSourceTopicRestartIntegrationTest) and saw there was some dependence on the empty group id.

Exception in thread "ktable-restore-from-source-2639ce74-b612-423d-aaff-96cc59653aa7-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: stream-thread [ktable-restore-from-source-2639ce74-b612-423d-aaff-96cc59653aa7-StreamThread-1] Failed to rebalance.
	at org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:922)
	at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:822)
	at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:777)
	at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:747)
Caused by: org.apache.kafka.common.errors.InvalidGroupIdException: To use the group management or offset commit APIs, you must provide a valid group.id in the consumer configuration.

cc @guozhangwang

vahidhashemian · 2018-11-14T17:43:46Z

Thanks @hachikuji for catching this.
I see this code in StreamsConfig.getRestoreConsumerConfigs:

        // no need to set group id for a restore consumer
        baseConsumerProps.remove(ConsumerConfig.GROUP_ID_CONFIG);

which now means setting the group id to null. I think it should be safe to replace it with

        // no need to use a specific group id for a restore consumer
        baseConsumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "");

Thoughts?
@guozhangwang

hachikuji · 2018-11-14T17:57:44Z

The exception is being raised from a call to unsubscribe. We currently allow this API to be used in order to remove all assigned partitions, so I think we should probably remove the group id validation from that API.

vahidhashemian · 2018-11-14T19:24:18Z

Sounds good, and seems to have no side effect with respect to existing unit tests. My initial thought was to use the empty string to keep the tests run with the same config as before. I'll submit a patch.

hachikuji

Thanks for the updates. Looks like the build is passing. Just two more comments and I think we can merge.

hachikuji · 2018-11-15T08:00:27Z

Shouldn't we handle the else case? I was thinking that if the user enables auto commit explicitly, but provides no group id, then we should raise a configuration exception.

What I had here was to let the consumer run and throw an exception when a commit is triggered (or even earlier when e.g. subscribe is called). But I think your suggestion makes more sense because it points directly to the root cause, i.e. the misconfiguration. Updated in the new patch.

hachikuji · 2018-11-15T08:02:07Z

nit: can we use Objects.requireNonNull instead?

Made this change too.

…IP-289) Improve the default group id behavior by * changing the default consumer group to null, where no offset commit or fetch, or group management operations are allowed * deprecate the use of empty (`""`) consumer group on the client

Also modified the logic in deciding whether auto-commit should be enable.

The validation caused Streams tests in `KTableSourceTopicRestartIntegrationTest` to fail.

vahidhashemian

@hachikuji Thanks for the feedback. Responded to your comment inline and submitted a new patch.

vahidhashemian · 2018-11-16T05:27:28Z

What I had here was to let the consumer run and throw an exception when a commit is triggered (or even earlier when e.g. subscribe is called). But I think your suggestion makes more sense because it points directly to the root cause, i.e. the misconfiguration. Updated in the new patch.

vahidhashemian · 2018-11-16T05:28:40Z

Made this change too.

hachikuji

LGTM. Pushed a trivial fix for the null check in AbstractCoordinator.

…IP-289) (apache#5877) Improve the default group id behavior by: * changing the default consumer group to null, where no offset commit or fetch, or group management operations are allowed * deprecating the use of empty (`""`) consumer group on the client Reviewers: Jason Gustafson <jason@confluent.io>

vahidhashemian force-pushed the KAFKA-6774 branch 2 times, most recently from 0b0dec0 to 73c9ba1 Compare November 4, 2018 07:03

hachikuji self-assigned this Nov 6, 2018

hachikuji reviewed Nov 7, 2018

View reviewed changes

vahidhashemian force-pushed the KAFKA-6774 branch from 73c9ba1 to 3e4b22c Compare November 7, 2018 06:53

vahidhashemian force-pushed the KAFKA-6774 branch from 3e4b22c to f775ccb Compare November 7, 2018 07:01

hachikuji reviewed Nov 7, 2018

View reviewed changes

vahidhashemian commented Nov 10, 2018

View reviewed changes

hachikuji reviewed Nov 10, 2018

View reviewed changes

vahidhashemian commented Nov 10, 2018

View reviewed changes

hachikuji reviewed Nov 12, 2018

View reviewed changes

vahidhashemian force-pushed the KAFKA-6774 branch from e5b9430 to db68a67 Compare November 13, 2018 06:49

hachikuji reviewed Nov 13, 2018

View reviewed changes

vahidhashemian commented Nov 14, 2018

View reviewed changes

vahidhashemian force-pushed the KAFKA-6774 branch from f6fec6c to 05a7594 Compare November 14, 2018 23:54

hachikuji reviewed Nov 15, 2018

View reviewed changes

vahidhashemian added 8 commits November 15, 2018 20:17

Avoid offset fetch when group id is null

f1a3a23

Added a couple of integration tests

ecb34f6

Also modified the logic in deciding whether auto-commit should be enable.

Minor unit test updates based on the recent feedback

bb18226

Disable coordinator construction when group id is null

d665452

Additional minor fixes

72c627f

Remove group id validation from unsubscribe() method

43e0083

The validation caused Streams tests in `KTableSourceTopicRestartIntegrationTest` to fail.

Throw a config exception for default group id with auto commit

7f5941e

vahidhashemian force-pushed the KAFKA-6774 branch from 05a7594 to 7f5941e Compare November 16, 2018 05:40

vahidhashemian commented Nov 16, 2018

View reviewed changes

Fix requireNonNull check

f8b1b9e

hachikuji approved these changes Nov 16, 2018

View reviewed changes

hachikuji merged commit c3e7d62 into apache:trunk Nov 16, 2018

Conversation

vahidhashemian commented Nov 4, 2018

Committer Checklist (excluded from commit message)

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

vahidhashemian commented Nov 7, 2018

Uh oh!

hachikuji left a comment

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!

vahidhashemian left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji left a comment

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!

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!

vahidhashemian left a comment

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!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

vahidhashemian commented Nov 13, 2018

Uh oh!

hachikuji left a comment

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!

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

hachikuji commented Nov 14, 2018 •

edited

Loading