You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2020/10/06 16:00:42 UTC
[GitHub] [kafka] lct45 opened a new pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
lct45 opened a new pull request #9383:
URL: https://github.com/apache/kafka/pull/9383
Add data to subscriptionUserData to make sure that it's different each time a consumer rejoins
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#discussion_r504110656
##########
File path: streams/src/test/java/org/apache/kafka/streams/tests/StreamsUpgradeTest.java
##########
@@ -158,7 +159,8 @@ public ByteBuffer subscriptionUserData(final Set<String> topics) {
LATEST_SUPPORTED_VERSION + 1,
taskManager.processId(),
userEndPoint(),
- taskManager.getTaskOffsetSums()
+ taskManager.getTaskOffsetSums(),
+ Bytes.EMPTY
Review comment:
We should do the same thing that we do in StreamsPartitionAssignor here, ie a changing byte. This is supposed to be like the same exact thing but for a "future version" of the partition assignor, to test version probing and general compatibility.
It's a bit annoying that you have to just copy the same exact stuff but that's just how it is for now
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#issuecomment-712403417
Cherry-picked to 2.6
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lct45 commented on a change in pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
lct45 commented on a change in pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#discussion_r501225751
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
##########
@@ -212,6 +213,7 @@ public void configure(final Map<String, ?> configs) {
rebalanceProtocol = assignorConfiguration.rebalanceProtocol();
taskAssignorSupplier = assignorConfiguration::taskAssignor;
assignmentListener = assignorConfiguration.assignmentListener();
+ uniqueField = usedSubscriptionMetadataVersion >= 8 ? new byte[1] : new byte[0];
Review comment:
This could be moved out to the `assignorConfiguration` class to match the other fields but the code is fairly straightforward so I left it here for now
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lct45 commented on pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
lct45 commented on pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#issuecomment-705688523
System tests: https://jenkins.confluent.io/job/system-test-kafka-branch-builder/4204/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lct45 commented on a change in pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
lct45 commented on a change in pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#discussion_r501956905
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
##########
@@ -483,7 +485,7 @@ public void testEagerSubscription() {
Collections.sort(subscription.topics());
assertEquals(asList("topic1", "topic2"), subscription.topics());
- final SubscriptionInfo info = getInfo(UUID_1, prevTasks, standbyTasks);
Review comment:
> goes back and forth between the two expected values
What do you mean by expected values? Toggling between the default of `0` and `1` after `partitionAssignor.subscriptionUserData` is called the first time?
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
##########
@@ -483,7 +485,7 @@ public void testEagerSubscription() {
Collections.sort(subscription.topics());
assertEquals(asList("topic1", "topic2"), subscription.topics());
- final SubscriptionInfo info = getInfo(UUID_1, prevTasks, standbyTasks);
Review comment:
> goes back and forth between the two expected values
What do you mean by expected values? Toggling between the default of `0` and `1` after `partitionAssignor.subscriptionUserData` is called the first time?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lct45 commented on pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
lct45 commented on pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#issuecomment-705688523
System tests: https://jenkins.confluent.io/job/system-test-kafka-branch-builder/4204/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#discussion_r501876646
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/SubscriptionInfoTest.java
##########
@@ -59,6 +60,7 @@
);
private final static String IGNORED_USER_ENDPOINT = "ignoredUserEndpoint:80";
+ private static final byte[] IGNORED_UNIQUE_FIELD = Bytes.EMPTY;
Review comment:
nit: let's use `new byte[1]` for this to make sure it's actually being ignored when it should be (since apparently it won't notice if you just pass in empty bytes for this field on a version < 8)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lct45 edited a comment on pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
lct45 edited a comment on pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#issuecomment-708708449
https://jenkins.confluent.io/job/system-test-kafka-branch-builder/4228/ system tests part 2
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#discussion_r501875691
##########
File path: streams/src/test/java/org/apache/kafka/streams/integration/HighAvailabilityTaskAssignorIntegrationTest.java
##########
@@ -286,7 +286,8 @@ private static Properties streamsProperties(final String appId,
mkEntry(StreamsConfig.PROBING_REBALANCE_INTERVAL_MS_CONFIG, "60000"),
mkEntry(StreamsConfig.InternalConfig.ASSIGNMENT_LISTENER, configuredAssignmentListener),
mkEntry(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, "100"),
- mkEntry(StreamsConfig.InternalConfig.INTERNAL_TASK_ASSIGNOR_CLASS, HighAvailabilityTaskAssignor.class.getName())
+ mkEntry(StreamsConfig.InternalConfig.INTERNAL_TASK_ASSIGNOR_CLASS, HighAvailabilityTaskAssignor.class.getName()),
+ mkEntry(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 40)
Review comment:
Can we add a comment here explaining why we set the thread count so high? I feel like we'll forget and be really confused when we stumble across this in the future.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#discussion_r504108760
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
##########
@@ -483,7 +485,7 @@ public void testEagerSubscription() {
Collections.sort(subscription.topics());
assertEquals(asList("topic1", "topic2"), subscription.topics());
- final SubscriptionInfo info = getInfo(UUID_1, prevTasks, standbyTasks);
Review comment:
Oh right duh I was thinking it was just a single bit but it's a byte. In that case we should have a test that verifies it goes from `0` to `1` to `2`, etc -- might be good to verify the behavior on overflow as well, if you call `subscriptionUserData` the max_value number of times
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lct45 commented on a change in pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
lct45 commented on a change in pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#discussion_r501329620
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
##########
@@ -212,6 +213,7 @@ public void configure(final Map<String, ?> configs) {
rebalanceProtocol = assignorConfiguration.rebalanceProtocol();
taskAssignorSupplier = assignorConfiguration::taskAssignor;
assignmentListener = assignorConfiguration.assignmentListener();
+ uniqueField = usedSubscriptionMetadataVersion >= 8 ? new byte[1] : new byte[0];
Review comment:
The autocode generator throws an error if `array.length != 0` and the version is < 8. It didn't seem to be a huge problem but there were some tests that used version 1 so initializing to `new byte[1]` threw errors
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#discussion_r501336666
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
##########
@@ -212,6 +213,7 @@ public void configure(final Map<String, ?> configs) {
rebalanceProtocol = assignorConfiguration.rebalanceProtocol();
taskAssignorSupplier = assignorConfiguration::taskAssignor;
assignmentListener = assignorConfiguration.assignmentListener();
+ uniqueField = usedSubscriptionMetadataVersion >= 8 ? new byte[1] : new byte[0];
Review comment:
That seems like an indication that the subscription is being encoded incorrectly. If the version is below 8, it shouldn't be trying to encode the `uniqueField` in the first place -- you shouldn't have to mimic that yourself by forcing it to be empty (and even if it is I worry there might be some additional bytes for metadata that would get stored for that field, when that field should not exist at all for versions < 8)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lct45 commented on pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
lct45 commented on pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#issuecomment-707127859
One test failing: testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#discussion_r501322300
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
##########
@@ -483,7 +485,7 @@ public void testEagerSubscription() {
Collections.sort(subscription.topics());
assertEquals(asList("topic1", "topic2"), subscription.topics());
- final SubscriptionInfo info = getInfo(UUID_1, prevTasks, standbyTasks);
Review comment:
Can you add a test that verifies that it goes back and forth between the two expected values when you call `partitionAssignor.subscriptionUserData` multiples times (let's say 3) -- also maybe add a verification that the `uniqueField` has a length of just 1
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
##########
@@ -234,15 +236,20 @@ public ByteBuffer subscriptionUserData(final Set<String> topics) {
// Adds the following information to subscription
// 1. Client UUID (a unique id assigned to an instance of KafkaStreams)
// 2. Map from task id to its overall lag
+ // 3. Unique Field to ensure a rebalance when a thread rejoins by forcing the user data to be different
handleRebalanceStart(topics);
+ if (usedSubscriptionMetadataVersion >= 8) {
Review comment:
Instead of hardcoded `8` all over, let's just define a constant for this similar to the `MIN_VERSION_OFFSET_SUM_SUBSCRIPTION ` in SubscriptionInfo
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
##########
@@ -234,15 +236,20 @@ public ByteBuffer subscriptionUserData(final Set<String> topics) {
// Adds the following information to subscription
// 1. Client UUID (a unique id assigned to an instance of KafkaStreams)
// 2. Map from task id to its overall lag
+ // 3. Unique Field to ensure a rebalance when a thread rejoins by forcing the user data to be different
handleRebalanceStart(topics);
+ if (usedSubscriptionMetadataVersion >= 8) {
Review comment:
If you want, I think it's also fine to just always flip the byte and not even check against the used subscription version.
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/SubscriptionInfoTest.java
##########
@@ -286,23 +296,23 @@ public void shouldEncodeAndDecodeSmallerLatestSupportedVersion() {
final int latestSupportedVersion = LATEST_SUPPORTED_VERSION - 1;
final SubscriptionInfo info =
- new SubscriptionInfo(usedVersion, latestSupportedVersion, UUID_1, "localhost:80", TASK_OFFSET_SUMS);
+ new SubscriptionInfo(usedVersion, latestSupportedVersion, UUID_1, "localhost:80", TASK_OFFSET_SUMS, IGNORED_UNIQUE_FIELD);
final SubscriptionInfo expectedInfo =
- new SubscriptionInfo(usedVersion, latestSupportedVersion, UUID_1, "localhost:80", TASK_OFFSET_SUMS);
+ new SubscriptionInfo(usedVersion, latestSupportedVersion, UUID_1, "localhost:80", TASK_OFFSET_SUMS, IGNORED_UNIQUE_FIELD);
assertEquals(expectedInfo, SubscriptionInfo.decode(info.encode()));
}
@Test
public void shouldEncodeAndDecodeVersion7() {
Review comment:
Can you add a test like this for the new version 8?
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
##########
@@ -212,6 +213,7 @@ public void configure(final Map<String, ?> configs) {
rebalanceProtocol = assignorConfiguration.rebalanceProtocol();
taskAssignorSupplier = assignorConfiguration::taskAssignor;
assignmentListener = assignorConfiguration.assignmentListener();
+ uniqueField = usedSubscriptionMetadataVersion >= 8 ? new byte[1] : new byte[0];
Review comment:
Also, now that I think about it, the `usedSubscriptionMetadataVersion` will only ever be >= 8 at this point. It might be set to something lower at some point later on, but it has to be at least 8 when the assignor is just being created/configured
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/SubscriptionInfoTest.java
##########
@@ -313,7 +323,8 @@ public void shouldReturnTaskOffsetSumsMapForDecodedSubscription() {
new SubscriptionInfo(MIN_VERSION_OFFSET_SUM_SUBSCRIPTION,
LATEST_SUPPORTED_VERSION, UUID_1,
"localhost:80",
- TASK_OFFSET_SUMS)
+ TASK_OFFSET_SUMS,
+ IGNORED_UNIQUE_FIELD)
.encode());
assertThat(info.taskOffsetSums(), is(TASK_OFFSET_SUMS));
}
Review comment:
I think we should also add a test to make sure that if you pass in a `uniqueField` to the SubscriptionInfo but the `usedVersion` is less than 8, that it does not actually encode this field.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
##########
@@ -212,6 +213,7 @@ public void configure(final Map<String, ?> configs) {
rebalanceProtocol = assignorConfiguration.rebalanceProtocol();
taskAssignorSupplier = assignorConfiguration::taskAssignor;
assignmentListener = assignorConfiguration.assignmentListener();
+ uniqueField = usedSubscriptionMetadataVersion >= 8 ? new byte[1] : new byte[0];
Review comment:
Sounds good. I think it's also fine to just initialize it to `new byte[1]` regardless of whether the version is high enough to actually need it or not. It's just a single byte, and simpler is better
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#discussion_r504111457
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
##########
@@ -483,7 +485,7 @@ public void testEagerSubscription() {
Collections.sort(subscription.topics());
assertEquals(asList("topic1", "topic2"), subscription.topics());
- final SubscriptionInfo info = getInfo(UUID_1, prevTasks, standbyTasks);
Review comment:
Oh, saw that you already added the test for `0` --> `1` --> `2` behavior. In that case can you just add two things to that test: verify the behavior on overflow, and verify that the length is always exactly one byte
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lct45 commented on a change in pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
lct45 commented on a change in pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#discussion_r502592684
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
##########
@@ -483,7 +485,7 @@ public void testEagerSubscription() {
Collections.sort(subscription.topics());
assertEquals(asList("topic1", "topic2"), subscription.topics());
- final SubscriptionInfo info = getInfo(UUID_1, prevTasks, standbyTasks);
Review comment:
I don't think it would go from `0` to `1` to `0` since the partition assigner always increases, so if you're using the same partition assignor it'll go from `0` to `1` to `2` etc
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lct45 edited a comment on pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
lct45 edited a comment on pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#issuecomment-709396656
> Why do we use a size-1 byte array instead of a single byte for this field?
@vvcephei There's a `RecordsFieldType` already created for `byte[]` but not one for `byte` so to get the `subscriptionInfo.json` to work I would've had to create a `RecordsFieldType` for `byte` and decided that leveraging the existing field type would be easier. If you're strongly opposed to this solution I can look into creating a new field type
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lct45 commented on a change in pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
lct45 commented on a change in pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#discussion_r501956905
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
##########
@@ -483,7 +485,7 @@ public void testEagerSubscription() {
Collections.sort(subscription.topics());
assertEquals(asList("topic1", "topic2"), subscription.topics());
- final SubscriptionInfo info = getInfo(UUID_1, prevTasks, standbyTasks);
Review comment:
> goes back and forth between the two expected values
What do you mean by expected values? Toggling between the default of `0` and `1` after `partitionAssignor.subscriptionUserData` is called the first time?
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
##########
@@ -483,7 +485,7 @@ public void testEagerSubscription() {
Collections.sort(subscription.topics());
assertEquals(asList("topic1", "topic2"), subscription.topics());
- final SubscriptionInfo info = getInfo(UUID_1, prevTasks, standbyTasks);
Review comment:
> goes back and forth between the two expected values
What do you mean by expected values? Toggling between the default of `0` and `1` after `partitionAssignor.subscriptionUserData` is called the first time?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lct45 commented on pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
lct45 commented on pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#issuecomment-708708449
https://jenkins.confluent.io/job/system-test-kafka-branch-builder/4226/ system tests part 2
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei merged pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
vvcephei merged pull request #9383:
URL: https://github.com/apache/kafka/pull/9383
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#issuecomment-712362984
Thanks for the PR, @lct45 !
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#discussion_r501875691
##########
File path: streams/src/test/java/org/apache/kafka/streams/integration/HighAvailabilityTaskAssignorIntegrationTest.java
##########
@@ -286,7 +286,8 @@ private static Properties streamsProperties(final String appId,
mkEntry(StreamsConfig.PROBING_REBALANCE_INTERVAL_MS_CONFIG, "60000"),
mkEntry(StreamsConfig.InternalConfig.ASSIGNMENT_LISTENER, configuredAssignmentListener),
mkEntry(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, "100"),
- mkEntry(StreamsConfig.InternalConfig.INTERNAL_TASK_ASSIGNOR_CLASS, HighAvailabilityTaskAssignor.class.getName())
+ mkEntry(StreamsConfig.InternalConfig.INTERNAL_TASK_ASSIGNOR_CLASS, HighAvailabilityTaskAssignor.class.getName()),
+ mkEntry(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 40)
Review comment:
Can we add a comment here explaining why we set the thread count so high? I feel like we'll forget and be really confused when we stumble across this in the future.
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/SubscriptionInfoTest.java
##########
@@ -59,6 +60,7 @@
);
private final static String IGNORED_USER_ENDPOINT = "ignoredUserEndpoint:80";
+ private static final byte[] IGNORED_UNIQUE_FIELD = Bytes.EMPTY;
Review comment:
nit: let's use `new byte[1]` for this to make sure it's actually being ignored when it should be (since apparently it won't notice if you just pass in empty bytes for this field on a version < 8)
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
##########
@@ -483,7 +485,7 @@ public void testEagerSubscription() {
Collections.sort(subscription.topics());
assertEquals(asList("topic1", "topic2"), subscription.topics());
- final SubscriptionInfo info = getInfo(UUID_1, prevTasks, standbyTasks);
Review comment:
Yeah, it should be `0` the first time you call it, then `1` the second time, and then back to `0` again on the third call
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lct45 edited a comment on pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
lct45 edited a comment on pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#issuecomment-708708449
https://jenkins.confluent.io/job/system-test-kafka-branch-builder/4227/ system tests part 2
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#discussion_r502052730
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
##########
@@ -483,7 +485,7 @@ public void testEagerSubscription() {
Collections.sort(subscription.topics());
assertEquals(asList("topic1", "topic2"), subscription.topics());
- final SubscriptionInfo info = getInfo(UUID_1, prevTasks, standbyTasks);
Review comment:
Yeah, it should be `0` the first time you call it, then `1` the second time, and then back to `0` again on the third call
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lct45 commented on pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
lct45 commented on pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#issuecomment-709396656
> Why do we use a size-1 byte array instead of a single byte for this field?
@vvcephei There's a `RecordsFieldType` already created for `byte[]` but not one for `byte` so to get the `subscriptionInfo.json` to work I would've had to create a `RecordsFieldType` for `byte` and decided that leveraging the existing field type would be easier. If you're strongly opposed to this solution I can look into creating a new field type
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on a change in pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
vvcephei commented on a change in pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#discussion_r506543174
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
##########
@@ -2023,13 +2022,13 @@ public void testUniqueField() {
configureDefaultPartitionAssignor();
final Set<String> topics = mkSet("input");
- assertEquals(1, partitionAssignor.uniqueField().length);
- assertEquals(0, partitionAssignor.uniqueField()[0]);
+ //assertEquals(1, partitionAssignor.uniqueField().length);
Review comment:
we probably ought to delete the commented-out assertions.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lct45 edited a comment on pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
lct45 edited a comment on pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#issuecomment-705688523
System tests: https://jenkins.confluent.io/job/system-test-kafka-branch-builder/4204/
Looks like AVE_MEMFD_CREATE failed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#issuecomment-712380123
Cherry-picked to 2.7 (cc @bbejeck )
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lct45 commented on pull request #9383: KAFKA-10455: Ensure that probing rebalances always occur
Posted by GitBox <gi...@apache.org>.
lct45 commented on pull request #9383:
URL: https://github.com/apache/kafka/pull/9383#issuecomment-704410260
@ableegoldman ready for initial review
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org