[
https://issues.apache.org/jira/browse/KAFKA-21409?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18136032#comment-18136032
]
Chia-Ping Tsai commented on KAFKA-21409:
----------------------------------------
{code:java}
@Timeout(120)
@ClusterTest(serverProperties = {
@ClusterConfigProperty(key =
CONSUMER_GROUP_ASSIGNMENT_INTERVAL_MS_CONFIG, value = "0")
})
public void
testAsyncConsumerRe2JPatternSwitchToNonMatchingRegexReassignsPartitions()
throws InterruptedException {
var groupId = "regex-switch-group";
Map<String, Object> config = Map.of(
GROUP_PROTOCOL_CONFIG,
GroupProtocol.CONSUMER.name().toLowerCase(Locale.ROOT),
GROUP_ID_CONFIG, groupId
);
try (Consumer<byte[], byte[]> consumer1 = cluster.consumer(config);
Consumer<byte[], byte[]> consumer2 = cluster.consumer(config)) {
// "topic" (2 partitions) is created by setup(). Both members
subscribe to a regex
// that matches it, so each one ends up with one partition.
var matching = new SubscriptionPattern("topic.*");
consumer1.subscribe(matching);
consumer2.subscribe(matching); Set<TopicPartition> all =
Set.of(
new TopicPartition(topic, 0),
new TopicPartition(topic, 1)
);
TestUtils.waitForCondition(() -> {
consumer1.poll(Duration.ofMillis(100));
consumer2.poll(Duration.ofMillis(100));
Set<TopicPartition> union = new
HashSet<>(consumer1.assignment());
union.addAll(consumer2.assignment());
return consumer1.assignment().size() == 1
&& consumer2.assignment().size() == 1
&& union.equals(all);
}, () -> "Expected one partition each, got " +
consumer1.assignment() + " and " + consumer2.assignment()); //
Member 2 switches, without leaving the group, to a regex that matches no topic.
// The client gives its partition back; the coordinator should hand
it to member 1.
consumer2.subscribe(new
SubscriptionPattern("nothing-matches-this.*")); // First, member 2
must give its partition back (the client treats an unresolved
// or non-matching regex as subscribing to nothing).
TestUtils.waitForCondition(() -> {
consumer1.poll(Duration.ofMillis(100));
consumer2.poll(Duration.ofMillis(100));
return consumer2.assignment().isEmpty();
}, 30_000L, () -> "Member 2 never released its partition: " +
consumer2.assignment()); // Then member 1 should receive it. The
regex refresh has a 10s minimum interval,
// so allow well over that.
TestUtils.waitForCondition(() -> {
consumer1.poll(Duration.ofMillis(100));
consumer2.poll(Duration.ofMillis(100));
return consumer1.assignment().equals(all);
}, 45_000L, () -> "Partition released by member 2 was never
reassigned to member 1. " +
"member1=" + consumer1.assignment() + " member2=" +
consumer2.assignment() +
" (group epoch was not bumped after the new regex resolved to
no topics)");
}
}
{code}
> Switching to a subscription regex that matches no topic doesn't trigger
> rebalance
> ---------------------------------------------------------------------------------
>
> Key: KAFKA-21409
> URL: https://issues.apache.org/jira/browse/KAFKA-21409
> Project: Kafka
> Issue Type: Bug
> Reporter: Chia-Ping Tsai
> Assignee: Chia-Ping Tsai
> Priority: Major
>
>
> {code:java}
> ResolvedRegularExpression newResolvedRegularExpression = entry.getValue();
> ResolvedRegularExpression oldResolvedRegularExpression = group
> .resolvedRegularExpression(regex)
> .orElse(ResolvedRegularExpression.EMPTY);
> if
> (!oldResolvedRegularExpression.topics().equals(newResolvedRegularExpression.topics()))
> {
> {code}
> Since ResolvedRegularExpression.EMPTY has no topics which is equal to
> "match-to-nothing", the condition won't get passed. The side effect is that
> the partitions won't be reassigned to other consumers.
>
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
