KAFKA-6145: KIP-441: Fix assignor config passthough by vvcephei · Pull Request #8716 · apache/kafka

vvcephei · 2020-05-22T16:47:44Z

Also fixes a system test by configuring the HATA to perform a one-shot balanced assignment

Committer Checklist (excluded from commit message)

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

vvcephei

Hey @cadonna and @ableegoldman , can you take a look at these fixes? I'm going to add a couple more tests, but I'd like to get your early feedback if you have time.

vvcephei · 2020-05-22T16:48:10Z

This is where we forgot to copy over the configs.

Can we add a note to AssignorConfiguration pointing to this for any new assignor-related configs that get added?

Do we automatically pass through the internal configs? I notice we don't copy over the task assignor class, or the new assignment listener callback I added for the integration tests. But both of them seem to get through

I know it's deprecated, but I think the PartitionGrouper config is missing as well.

Yes, we copy over the other internal configs in org.apache.kafka.streams.processor.internals.StreamThread#create

I'll add your listener to the config test. I'm not sure about PartitionGrouper.

Turns out PartitionGrouper does not get copied over. I'll create a jira to track this, so that we don't have to get sidetracked in this PR with the question of what we should do about this deprecated config.

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

we automatically pass through any config that isn't registered

I have to say, this seems totally backwards to me. So basically we just happen to pass in any number of configs that we may or may not need, but will split out specific configs that we do need unless explicitly told to include them? I understand the custom configs motivation, but then why not just pass through all the configs?

What if I wanted to access the value of one of my registered Streams configs in my plugged-in component? I'd have to add the same value a second time, but with an unregistered config name. Huh?

Regarding KAFKA-10046, in current trunk we already have some logic that assumes the default partition grouper is always used, so I'd suggest we just bite the bullet and remove it in 2.6.

Thanks, @guozhangwang

@ableegoldman , it seems backwards to me also.

vvcephei · 2020-05-22T16:49:03Z

I added this constraint to mirror the constraint we already apply in StreamConfig. It's not critical, but I was disappointed that I had written a bunch of tests that included a technically invalid configuration.

Should we add the same for all the other configs?

IMO, we should check the limits for all configs. However, I am not sure if we should check probingRebalanceIntervalMs to be >= 60 * 1000L (as we do in StreamsConfig) or just `>= 0.

We could; to be fair, the config parser already checks the limits, so it's not necessary for production code.

It only comes up when we manually create these internal objects for tests. I added this particular bound specifically because I had previously passed in what I thought was a dummy value, which turned out to be a misconfiguration that actually affected the behavior I was testing.

Offhand, it doesn't seem like the same thing would happen with probingRebalanceIntervalMs, so it doesn't seem like the check here would have the same benefit; WDYT?

Nevermind; I refactored the class to use the same config validation for passed-in arguments.

vvcephei · 2020-05-22T16:49:19Z

I found this useful while debugging the system test.

vvcephei · 2020-05-22T16:50:14Z

Since we can't have zero warmups, we don't need this condition (that I added in #8696)

vvcephei · 2020-05-22T16:51:18Z

Fixing a double-space we were printing when there was a followup. (It would say with followup)

vvcephei · 2020-05-22T17:05:24Z

Expanding DeMorgan's law at @cadonna 's request (which I also appreciated).

vvcephei · 2020-05-22T17:10:30Z

The diff is misaligned. I removed shouldBeStickyForActiveAndStandbyTasksEvenIfNoWarmups and added shouldSkipWarmupsWhenAcceptableLagIsMax.

vvcephei · 2020-05-22T17:10:50Z

All these tests erroneously set "max warmups" to zero.

vvcephei · 2020-05-22T17:11:19Z

It was handy to be able to see the used config file while debugging.

vvcephei · 2020-05-22T17:11:58Z

Added this configuration to fix the flaky StreamsOptimizedTest.test_upgrade_optimized_topology

vvcephei · 2020-05-22T18:43:06Z

FWIW:

================================================================================
SESSION REPORT (ALL TESTS)
ducktape version: 0.7.7
session_id:       2020-05-22--007
run time:         29 minutes 9.205 seconds
tests run:        10
passed:           10
failed:           0
ignored:          0
================================================================================
test_id:    kafkatest.tests.streams.streams_optimized_test.StreamsOptimizedTest.test_upgrade_optimized_topology
status:     PASS
run time:   2 minutes 32.957 seconds
--------------------------------------------------------------------------------
test_id:    kafkatest.tests.streams.streams_optimized_test.StreamsOptimizedTest.test_upgrade_optimized_topology
status:     PASS
run time:   2 minutes 31.512 seconds
--------------------------------------------------------------------------------
test_id:    kafkatest.tests.streams.streams_optimized_test.StreamsOptimizedTest.test_upgrade_optimized_topology
status:     PASS
run time:   2 minutes 56.126 seconds
--------------------------------------------------------------------------------
test_id:    kafkatest.tests.streams.streams_optimized_test.StreamsOptimizedTest.test_upgrade_optimized_topology
status:     PASS
run time:   2 minutes 55.289 seconds
--------------------------------------------------------------------------------
test_id:    kafkatest.tests.streams.streams_optimized_test.StreamsOptimizedTest.test_upgrade_optimized_topology
status:     PASS
run time:   2 minutes 52.812 seconds
--------------------------------------------------------------------------------
test_id:    kafkatest.tests.streams.streams_optimized_test.StreamsOptimizedTest.test_upgrade_optimized_topology
status:     PASS
run time:   3 minutes 4.072 seconds
--------------------------------------------------------------------------------
test_id:    kafkatest.tests.streams.streams_optimized_test.StreamsOptimizedTest.test_upgrade_optimized_topology
status:     PASS
run time:   3 minutes 3.525 seconds
--------------------------------------------------------------------------------
test_id:    kafkatest.tests.streams.streams_optimized_test.StreamsOptimizedTest.test_upgrade_optimized_topology
status:     PASS
run time:   3 minutes 4.718 seconds
--------------------------------------------------------------------------------
test_id:    kafkatest.tests.streams.streams_optimized_test.StreamsOptimizedTest.test_upgrade_optimized_topology
status:     PASS
run time:   3 minutes 3.213 seconds
--------------------------------------------------------------------------------
test_id:    kafkatest.tests.streams.streams_optimized_test.StreamsOptimizedTest.test_upgrade_optimized_topology
status:     PASS
run time:   3 minutes 4.576 seconds
--------------------------------------------------------------------------------

vvcephei · 2020-05-22T18:43:46Z

Previously, it would always fail for me within the first one to three tests, the first one more often than not.

vvcephei · 2020-05-22T19:26:35Z

Added this test, and verified that it does indeed fail on trunk for the expected reason that the new configs were ignored and defaults were substituted instead.

ableegoldman · 2020-05-22T23:47:39Z

Fair enough. I don't think it was meant as a cost saving thing, just to make it easier to understand when something did or did not have caught-up clients. If you find this logic easier to follow, go for it

ableegoldman · 2020-05-22T23:50:32Z

Nice catch! One nit is that "active" alone is not sufficient for being considered caught-up. Can we rename the active condition to running or activeRunning, etc?

ableegoldman · 2020-05-22T23:53:56Z

Do we automatically pass through the internal configs? I notice we don't copy over the task assignor class, or the new assignment listener callback I added for the integration tests. But both of them seem to get through

ableegoldman · 2020-05-22T23:58:45Z

I know it's deprecated, but I think the PartitionGrouper config is missing as well.

ableegoldman · 2020-05-23T00:09:05Z

Should we move the handful of AssignorConfiguration related tests from StreamsPartitionAssignorTest to here?

Yes, please!

Ah, unfortunately, that test relies on mocking package-private fields from another package. I'll just leave it alone for now.

ableegoldman · 2020-05-23T00:29:57Z

Trying to avoid piling on even more unrelated questions to the thread below, but there's another config that would need to be passed in, the admin client timeout.
That said, can we just remove it? It only gets the timeout the admin is configured with anyway

Ack, I'll take a look at it.

It does wind up "surviving" into the config that we use in the assignor, but I'm not sure if it's on purpose or just lucky.

I'll postpone any existential questions, and just add it to the regression test.

Ah, it was also tested here: org.apache.kafka.streams.StreamsConfigTest#consumerConfigShouldContainAdminClientConfigsForRetriesAndRetryBackOffMsWithAdminPrefix

I'll open a separate PR to remove the extra timeout config

cadonna

Thank you for the PR, @vvcephei !

Here my feedback!

cadonna · 2020-05-25T12:29:27Z

req: Please add verifications to StreamsConfigTest#consumerConfigMustContainStreamPartitionAssignorConfig()

cadonna · 2020-05-25T12:40:36Z

IMO, we should check the limits for all configs. However, I am not sure if we should check probingRebalanceIntervalMs to be >= 60 * 1000L (as we do in StreamsConfig) or just `>= 0.

cadonna · 2020-05-25T13:02:08Z

Nice catch indeed! I agree with @ableegoldman about the renaming. I am in favour of activeRunning or activeAndRunning.

cadonna · 2020-05-25T13:12:09Z

Yes, please!

vvcephei · 2020-05-26T23:28:14Z

Thanks for the reviews, @ableegoldman and @cadonna ! I believe I've addressed all your feedback.

vvcephei · 2020-05-27T00:12:46Z

This java 14 failure is strange:

00:06:34.561 > Task :streams:compileTestJava
00:06:34.561 /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk14-scala2.13/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressScenarioTest.java:495: error: incompatible types: inference variable K#1 has incompatible bounds
00:06:34.561             verify(
00:06:34.562                   ^
00:06:34.562     equality constraints: V#1,V#1,V#1,V#1,V#1,V#1,V#1,K#2,K#3,String,K#1,K#1,K#1,K#1,K#1,K#1
00:06:34.562     lower bounds: Long,String
00:06:34.562   where K#1,V#1,K#2,V#2,K#3,V#3 are type-variables:
00:06:34.562     K#1 extends Object declared in class KeyValueTimestamp
00:06:34.562     V#1 extends Object declared in class KeyValueTimestamp
00:06:34.562     K#2 extends Object declared in method <K#2,V#2>verify(List<TestRecord<K#2,V#2>>,List<KeyValueTimestamp<K#2,V#2>>)
00:06:34.711     V#2 extends Object declared in method <K#2,V#2>verify(List<TestRecord<K#2,V#2>>,List<KeyValueTimestamp<K#2,V#2>>)
00:06:34.711     K#3 extends Object declared in method <K#3,V#3>drainProducerRecords(TopologyTestDriver,String,Deserializer<K#3>,Deserializer<V#3>)
00:06:34.711     V#3 extends Object declared in method <K#3,V#3>drainProducerRecords(TopologyTestDriver,String,Deserializer<K#3>,Deserializer<V#3>)
00:06:40.558 1 error

I'm going to try again, and see what we see.

vvcephei · 2020-05-27T00:12:52Z

Retest this please

vvcephei · 2020-05-27T00:13:01Z

Retest this please

ableegoldman · 2020-05-27T00:28:59Z

I think this comment got lost in a discussion thread, but can we add a note to AssignorConfiguration pointing out that any Streams configs added here will need to be explicitly passed through? It seems like it's too easy to fall into this same trap again

Ah, right. Sure thing!

vvcephei · 2020-05-27T14:34:45Z

The test failures were the result of extra tests that were added on trunk after this branch. The branch builder does a merge with trunk before running the tests. I'm rebasing and fixing the tests.

vvcephei · 2020-05-27T17:28:44Z

The only failures were:
org.apache.kafka.streams.integration.EosBetaUpgradeIntegrationTest.shouldUpgradeFromEosAlphaToEosBeta[false]
org.apache.kafka.streams.integration.EosBetaUpgradeIntegrationTest.shouldUpgradeFromEosAlphaToEosBeta[true]

ableegoldman

One comment on the TaskAssignorIntegrationTest but feel free to merge as-is

ableegoldman · 2020-05-27T17:43:43Z

        public static final String TIME = "__time__";

        // This is settable in the main Streams config, but it's a private API for testing
-        public static final String ASSIGNMENT_LISTENER = "__asignment.listener__";


🤦‍♀️

ableegoldman · 2020-05-27T17:46:16Z

I'll open a separate PR to remove the extra timeout config

ableegoldman · 2020-05-27T17:51:43Z

+                mkEntry(StreamsConfig.ACCEPTABLE_RECOVERY_LAG_CONFIG, "6"),
+                mkEntry(StreamsConfig.MAX_WARMUP_REPLICAS_CONFIG, "7"),
+                mkEntry(StreamsConfig.PROBING_REBALANCE_INTERVAL_MS_CONFIG, "480000"),
+                mkEntry(StreamsConfig.InternalConfig.ASSIGNMENT_LISTENER, configuredAssignmentListener),


Should we also validate that the task assignor gets passed through? We could even pass in a custom assignor and use that to verify the correct assignor configs got passed through.

Of course, reflection black magic-ry is just more fun 🙂

Good idea; I'm not sure why I didn't think to do this already. With my newfound understanding of this config translation logic, I'm confident that it gets copied over right now, because it's not a registered config, but a regression test would be nice.

I'll quickly follow up with a separate PR so that I can merge this one.

Sounds good. Thanks for the fix!

vvcephei added the streams label May 22, 2020

vvcephei commented May 22, 2020

View reviewed changes

ableegoldman reviewed May 23, 2020

View reviewed changes

cadonna reviewed May 25, 2020

View reviewed changes

ableegoldman reviewed May 27, 2020

View reviewed changes

KAFKA-6145: KIP-441: Fix assignor config passthough

3c8426c

fix test

94194f2

ableegoldman approved these changes May 27, 2020

View reviewed changes

vvcephei merged commit 2cff1fa into apache:trunk May 27, 2020

vvcephei deleted the kafka-6145-one-shot-balance branch May 27, 2020 18:50

vvcephei mentioned this pull request May 28, 2020

MINOR: regression test for task assignor config #8743

Merged

3 tasks

mjsax added the kip Requires or implements a KIP label Jun 12, 2020

Conversation

vvcephei commented May 22, 2020

Committer Checklist (excluded from commit message)

Uh oh!

vvcephei left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ableegoldman May 22, 2020 • 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!

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!

vvcephei May 22, 2020 • 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!

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!

vvcephei commented May 22, 2020

Uh oh!

vvcephei commented May 22, 2020

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!

vvcephei May 26, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

ableegoldman May 22, 2020 •

edited

Loading

vvcephei May 22, 2020 •

edited

Loading

vvcephei May 26, 2020 •

edited

Loading