KAFKA-4633; Always using regex pattern subscription in StreamThread 1. In StreamThread, always use subscribe(Pattern, ..) function in order to avoid sending MetadataRequest with specific topic names and cause brokers to possibly auto-create subscribed topics; the pattern is generated as "topic-1|topic-2..|topic-n".
2. In ConsumerCoordinator, let the leader to refresh its metadata if the generated assignment contains some topics that is not contained in the subscribed topics; also in SubscriptionState, modified the verification for regex subscription to against the regex pattern instead of the matched topics since the returned assignment may contain some topics not yet created when joining the group but existed after the rebalance; also modified some unit tests in `KafkaConsumerTest` to accommodate the above changes. 3. Minor cleanup: changed String[] to List<String> to avoid overloaded functions. 4. Minor cleanup: enforced strong typing in SinkNodeFactory and removed unnecessary unchecked tags. 5. Minor cleanup: augmented unit test error message and fixed a potential transient failure in KafkaStreamTest. Author: Guozhang Wang <[email protected]> Reviewers: Damian Guy <[email protected]>, Matthias J. Sax <[email protected]>, Jason Gustafson <[email protected]> Closes #2379 from guozhangwang/K4633-regex-pattern Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/3400d0c3 Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/3400d0c3 Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/3400d0c3 Branch: refs/heads/trunk Commit: 3400d0c3cc07d5d3136bf9a19142b36cca93f92d Parents: 62206de Author: Guozhang Wang <[email protected]> Authored: Mon Jan 23 10:45:45 2017 -0800 Committer: Jason Gustafson <[email protected]> Committed: Mon Jan 23 10:45:45 2017 -0800 ---------------------------------------------------------------------- .../consumer/internals/ConsumerCoordinator.java | 59 ++- .../consumer/internals/SubscriptionState.java | 13 +- .../clients/consumer/KafkaConsumerTest.java | 6 +- .../internals/ConsumerCoordinatorTest.java | 410 ++++++++++++------- .../internals/SubscriptionStateTest.java | 53 ++- .../kstream/internals/AbstractStream.java | 6 - .../streams/processor/TopologyBuilder.java | 182 ++++---- .../streams/processor/internals/SinkNode.java | 4 +- .../streams/processor/internals/SourceNode.java | 9 +- .../processor/internals/StreamThread.java | 13 +- .../internals/StreamsMetadataState.java | 16 +- .../apache/kafka/streams/KafkaStreamsTest.java | 2 + .../integration/KStreamRepartitionJoinTest.java | 6 +- .../integration/utils/EmbeddedKafkaCluster.java | 2 +- .../integration/utils/IntegrationTestUtils.java | 4 +- .../streams/kstream/KStreamBuilderTest.java | 12 +- .../streams/processor/TopologyBuilderTest.java | 21 +- .../SourceNodeRecordDeserializerTest.java | 4 +- .../org/apache/kafka/test/MockSourceNode.java | 3 +- 19 files changed, 478 insertions(+), 347 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kafka/blob/3400d0c3/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java index 7c463d1..4c54a8f 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java @@ -216,6 +216,32 @@ public final class ConsumerCoordinator extends AbstractCoordinator { // update partition assignment subscriptions.assignFromSubscribed(assignment.partitions()); + // check if the assignment contains some topics that were not in the original + // subscription, if yes we will obey what leader has decided and add these topics + // into the subscriptions as long as they still match the subscribed pattern + // + // TODO this part of the logic should be removed once we allow regex on leader assign + Set<String> addedTopics = new HashSet<>(); + for (TopicPartition tp : subscriptions.assignedPartitions()) { + if (!joinedSubscription.contains(tp.topic())) + addedTopics.add(tp.topic()); + } + + if (!addedTopics.isEmpty()) { + Set<String> newSubscription = new HashSet<>(subscriptions.subscription()); + Set<String> newJoinedSubscription = new HashSet<>(joinedSubscription); + newSubscription.addAll(addedTopics); + newJoinedSubscription.addAll(addedTopics); + + this.subscriptions.subscribeFromPattern(newSubscription); + this.joinedSubscription = newJoinedSubscription; + } + + // update the metadata and enforce a refresh to make sure the fetcher can start + // fetching data in the next iteration + this.metadata.setTopics(subscriptions.groupSubscription()); + client.ensureFreshMetadata(); + // give the assignor a chance to update internal state based on the received assignment assignor.onAssignment(assignment); @@ -307,13 +333,44 @@ public final class ConsumerCoordinator extends AbstractCoordinator { client.ensureFreshMetadata(); isLeader = true; - assignmentSnapshot = metadataSnapshot; log.debug("Performing assignment for group {} using strategy {} with subscriptions {}", groupId, assignor.name(), subscriptions); Map<String, Assignment> assignment = assignor.assign(metadata.fetch(), subscriptions); + // user-customized assignor may have created some topics that are not in the subscription list + // and assign their partitions to the members; in this case we would like to update the leader's + // own metadata with the newly added topics so that it will not trigger a subsequent rebalance + // when these topics gets updated from metadata refresh. + // + // TODO: this is a hack and not something we want to support long-term unless we push regex into the protocol + // we may need to modify the PartitionAssingor API to better support this case. + Set<String> assignedTopics = new HashSet<>(); + for (Assignment assigned : assignment.values()) { + for (TopicPartition tp : assigned.partitions()) + assignedTopics.add(tp.topic()); + } + + if (!assignedTopics.containsAll(allSubscribedTopics)) { + Set<String> notAssignedTopics = new HashSet<>(allSubscribedTopics); + notAssignedTopics.removeAll(assignedTopics); + log.warn("The following subscribed topics are not assigned to any members in the group {} : {} ", groupId, notAssignedTopics); + } + + if (!allSubscribedTopics.containsAll(assignedTopics)) { + Set<String> newlyAddedTopics = new HashSet<>(assignedTopics); + newlyAddedTopics.removeAll(allSubscribedTopics); + log.info("The following not-subscribed topics are assigned to group {}, and their metadata will be fetched from the brokers : {}", groupId, newlyAddedTopics); + + allSubscribedTopics.addAll(assignedTopics); + this.subscriptions.groupSubscribe(allSubscribedTopics); + metadata.setTopics(this.subscriptions.groupSubscription()); + client.ensureFreshMetadata(); + } + + assignmentSnapshot = metadataSnapshot; + log.debug("Finished assignment for group {}: {}", groupId, assignment); Map<String, ByteBuffer> groupAssignment = new HashMap<>(); http://git-wip-us.apache.org/repos/asf/kafka/blob/3400d0c3/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java index 9e496ff..1a2a7ee 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java @@ -183,9 +183,16 @@ public class SubscriptionState { Set<TopicPartition> newAssignment = new HashSet<>(assignments); removeAllLagSensors(newAssignment); - for (TopicPartition tp : assignments) - if (!this.subscription.contains(tp.topic())) - throw new IllegalArgumentException("Assigned partition " + tp + " for non-subscribed topic."); + if (this.subscribedPattern != null) { + for (TopicPartition tp : assignments) { + if (!this.subscribedPattern.matcher(tp.topic()).matches()) + throw new IllegalArgumentException("Assigned partition " + tp + " for non-subscribed topic regex pattern; subscription pattern is " + this.subscribedPattern); + } + } else { + for (TopicPartition tp : assignments) + if (!this.subscription.contains(tp.topic())) + throw new IllegalArgumentException("Assigned partition " + tp + " for non-subscribed topic; subscription is " + this.subscription); + } // after rebalancing, we always reinitialize the assignment value this.assignment.set(partitionToStateMap(assignments)); http://git-wip-us.apache.org/repos/asf/kafka/blob/3400d0c3/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java ---------------------------------------------------------------------- diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java index 3e8310d..ac88ce9 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java @@ -366,7 +366,7 @@ public class KafkaConsumerTest { Metadata metadata = new Metadata(0, Long.MAX_VALUE); metadata.update(cluster, time.milliseconds()); - MockClient client = new MockClient(time); + MockClient client = new MockClient(time, metadata); client.setNode(node); PartitionAssignor assignor = new RoundRobinAssignor(); @@ -1228,7 +1228,7 @@ public class KafkaConsumerTest { Metadata metadata = new Metadata(0, Long.MAX_VALUE); metadata.update(cluster, time.milliseconds()); - MockClient client = new MockClient(time); + MockClient client = new MockClient(time, metadata); client.setNode(node); PartitionAssignor assignor = new RoundRobinAssignor(); @@ -1238,6 +1238,8 @@ public class KafkaConsumerTest { consumer.subscribe(Arrays.asList(topic), getConsumerRebalanceListener(consumer)); Node coordinator = prepareRebalance(client, node, assignor, Arrays.asList(tp0), null); + client.prepareMetadataUpdate(cluster); + // Poll with responses client.prepareResponseFrom(fetchResponse(tp0, 0, 1), node); client.prepareResponseFrom(fetchResponse(tp0, 1, 0), node); http://git-wip-us.apache.org/repos/asf/kafka/blob/3400d0c3/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java ---------------------------------------------------------------------- diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java index ee6afe1..3c4dd2d 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java @@ -85,9 +85,11 @@ import static org.junit.Assert.fail; public class ConsumerCoordinatorTest { - private String topicName = "test"; + private String topic1 = "test1"; + private String topic2 = "test2"; private String groupId = "test-group"; - private TopicPartition tp = new TopicPartition(topicName, 0); + private TopicPartition t1p = new TopicPartition(topic1, 0); + private TopicPartition t2p = new TopicPartition(topic2, 0); private int rebalanceTimeoutMs = 60000; private int sessionTimeoutMs = 10000; private int heartbeatIntervalMs = 5000; @@ -98,7 +100,12 @@ public class ConsumerCoordinatorTest { private List<PartitionAssignor> assignors = Collections.<PartitionAssignor>singletonList(partitionAssignor); private MockTime time; private MockClient client; - private Cluster cluster = TestUtils.singletonCluster(topicName, 1); + private Cluster cluster = TestUtils.clusterWith(1, new HashMap<String, Integer>() { + { + put(topic1, 1); + put(topic2, 1); + } + }); private Node node = cluster.nodes().get(0); private SubscriptionState subscriptions; private Metadata metadata; @@ -112,10 +119,10 @@ public class ConsumerCoordinatorTest { @Before public void setup() { this.time = new MockTime(); - this.client = new MockClient(time); this.subscriptions = new SubscriptionState(OffsetResetStrategy.EARLIEST, metrics); this.metadata = new Metadata(0, Long.MAX_VALUE); this.metadata.update(cluster, time.milliseconds()); + this.client = new MockClient(time, metadata); this.consumerClient = new ConsumerNetworkClient(client, metadata, time, 100, 1000); this.metrics = new Metrics(time); this.rebalanceListener = new MockRebalanceListener(); @@ -157,7 +164,7 @@ public class ConsumerCoordinatorTest { @Test(expected = GroupAuthorizationException.class) public void testGroupReadUnauthorized() { - subscriptions.subscribe(singleton(topicName), rebalanceListener); + subscriptions.subscribe(singleton(topic1), rebalanceListener); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); @@ -215,8 +222,8 @@ public class ConsumerCoordinatorTest { coordinator.ensureCoordinatorReady(); // illegal_generation will cause re-partition - subscriptions.subscribe(singleton(topicName), rebalanceListener); - subscriptions.assignFromSubscribed(Collections.singletonList(tp)); + subscriptions.subscribe(singleton(topic1), rebalanceListener); + subscriptions.assignFromSubscribed(Collections.singletonList(t1p)); time.sleep(sessionTimeoutMs); RequestFuture<Void> future = coordinator.sendHeartbeatRequest(); // should send out the heartbeat @@ -239,8 +246,8 @@ public class ConsumerCoordinatorTest { coordinator.ensureCoordinatorReady(); // illegal_generation will cause re-partition - subscriptions.subscribe(singleton(topicName), rebalanceListener); - subscriptions.assignFromSubscribed(Collections.singletonList(tp)); + subscriptions.subscribe(singleton(topic1), rebalanceListener); + subscriptions.assignFromSubscribed(Collections.singletonList(t1p)); time.sleep(sessionTimeoutMs); RequestFuture<Void> future = coordinator.sendHeartbeatRequest(); // should send out the heartbeat @@ -282,10 +289,10 @@ public class ConsumerCoordinatorTest { public void testJoinGroupInvalidGroupId() { final String consumerId = "leader"; - subscriptions.subscribe(singleton(topicName), rebalanceListener); + subscriptions.subscribe(singleton(topic1), rebalanceListener); // ensure metadata is up-to-date for leader - metadata.setTopics(singletonList(topicName)); + metadata.setTopics(singletonList(topic1)); metadata.update(cluster, time.milliseconds()); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); @@ -300,18 +307,18 @@ public class ConsumerCoordinatorTest { public void testNormalJoinGroupLeader() { final String consumerId = "leader"; - subscriptions.subscribe(singleton(topicName), rebalanceListener); + subscriptions.subscribe(singleton(topic1), rebalanceListener); // ensure metadata is up-to-date for leader - metadata.setTopics(singletonList(topicName)); + metadata.setTopics(singletonList(topic1)); metadata.update(cluster, time.milliseconds()); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); // normal join group - Map<String, List<String>> memberSubscriptions = Collections.singletonMap(consumerId, singletonList(topicName)); - partitionAssignor.prepare(Collections.singletonMap(consumerId, singletonList(tp))); + Map<String, List<String>> memberSubscriptions = Collections.singletonMap(consumerId, singletonList(topic1)); + partitionAssignor.prepare(Collections.singletonMap(consumerId, singletonList(t1p))); client.prepareResponse(joinGroupLeaderResponse(1, consumerId, memberSubscriptions, Errors.NONE.code())); client.prepareResponse(new MockClient.RequestMatcher() { @@ -322,37 +329,79 @@ public class ConsumerCoordinatorTest { sync.generationId() == 1 && sync.groupAssignment().containsKey(consumerId); } - }, syncGroupResponse(singletonList(tp), Errors.NONE.code())); + }, syncGroupResponse(singletonList(t1p), Errors.NONE.code())); coordinator.poll(time.milliseconds()); assertFalse(coordinator.needRejoin()); - assertEquals(singleton(tp), subscriptions.assignedPartitions()); + assertEquals(singleton(t1p), subscriptions.assignedPartitions()); + assertEquals(singleton(topic1), subscriptions.groupSubscription()); assertEquals(1, rebalanceListener.revokedCount); assertEquals(Collections.emptySet(), rebalanceListener.revoked); assertEquals(1, rebalanceListener.assignedCount); - assertEquals(singleton(tp), rebalanceListener.assigned); + assertEquals(singleton(t1p), rebalanceListener.assigned); + } + + @Test + public void testPatternJoinGroupLeader() { + final String consumerId = "leader"; + + subscriptions.subscribe(Pattern.compile("test.*"), rebalanceListener); + + // partially update the metadata with one topic first, + // let the leader to refresh metadata during assignment + metadata.setTopics(singletonList(topic1)); + metadata.update(TestUtils.singletonCluster(topic1, 1), time.milliseconds()); + + client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); + coordinator.ensureCoordinatorReady(); + + // normal join group + Map<String, List<String>> memberSubscriptions = Collections.singletonMap(consumerId, singletonList(topic1)); + partitionAssignor.prepare(Collections.singletonMap(consumerId, Arrays.asList(t1p, t2p))); + + client.prepareResponse(joinGroupLeaderResponse(1, consumerId, memberSubscriptions, Errors.NONE.code())); + client.prepareResponse(new MockClient.RequestMatcher() { + @Override + public boolean matches(AbstractRequest body) { + SyncGroupRequest sync = (SyncGroupRequest) body; + return sync.memberId().equals(consumerId) && + sync.generationId() == 1 && + sync.groupAssignment().containsKey(consumerId); + } + }, syncGroupResponse(Arrays.asList(t1p, t2p), Errors.NONE.code())); + // expect client to force updating the metadata, if yes gives it both topics + client.prepareMetadataUpdate(cluster); + + coordinator.poll(time.milliseconds()); + + assertFalse(coordinator.needRejoin()); + assertEquals(2, subscriptions.assignedPartitions().size()); + assertEquals(2, subscriptions.groupSubscription().size()); + assertEquals(2, subscriptions.subscription().size()); + assertEquals(1, rebalanceListener.revokedCount); + assertEquals(Collections.emptySet(), rebalanceListener.revoked); + assertEquals(1, rebalanceListener.assignedCount); + assertEquals(2, rebalanceListener.assigned.size()); } @Test public void testMetadataRefreshDuringRebalance() { final String consumerId = "leader"; - final String otherTopicName = "otherTopic"; - TopicPartition otherPartition = new TopicPartition(otherTopicName, 0); subscriptions.subscribe(Pattern.compile(".*"), rebalanceListener); metadata.needMetadataForAllTopics(true); - metadata.update(cluster, time.milliseconds()); + metadata.update(TestUtils.singletonCluster(topic1, 1), time.milliseconds()); - assertEquals(singleton(topicName), subscriptions.subscription()); + assertEquals(singleton(topic1), subscriptions.subscription()); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); - Map<String, List<String>> initialSubscription = singletonMap(consumerId, singletonList(topicName)); - partitionAssignor.prepare(singletonMap(consumerId, singletonList(tp))); + Map<String, List<String>> initialSubscription = singletonMap(consumerId, singletonList(topic1)); + partitionAssignor.prepare(singletonMap(consumerId, singletonList(t1p))); // the metadata will be updated in flight with a new topic added - final List<String> updatedSubscription = Arrays.asList(topicName, otherTopicName); + final List<String> updatedSubscription = Arrays.asList(topic1, topic2); final Set<String> updatedSubscriptionSet = new HashSet<>(updatedSubscription); client.prepareResponse(joinGroupLeaderResponse(1, consumerId, initialSubscription, Errors.NONE.code())); @@ -365,12 +414,12 @@ public class ConsumerCoordinatorTest { metadata.update(TestUtils.clusterWith(1, updatedPartitions), time.milliseconds()); return true; } - }, syncGroupResponse(singletonList(tp), Errors.NONE.code())); + }, syncGroupResponse(singletonList(t1p), Errors.NONE.code())); - List<TopicPartition> newAssignment = Arrays.asList(tp, otherPartition); + List<TopicPartition> newAssignment = Arrays.asList(t1p, t2p); Set<TopicPartition> newAssignmentSet = new HashSet<>(newAssignment); - Map<String, List<String>> updatedSubscriptions = singletonMap(consumerId, Arrays.asList(topicName, otherTopicName)); + Map<String, List<String>> updatedSubscriptions = singletonMap(consumerId, Arrays.asList(topic1, topic2)); partitionAssignor.prepare(singletonMap(consumerId, newAssignment)); // we expect to see a second rebalance with the new-found topics @@ -392,7 +441,7 @@ public class ConsumerCoordinatorTest { assertEquals(updatedSubscriptionSet, subscriptions.subscription()); assertEquals(newAssignmentSet, subscriptions.assignedPartitions()); assertEquals(2, rebalanceListener.revokedCount); - assertEquals(singleton(tp), rebalanceListener.revoked); + assertEquals(singleton(t1p), rebalanceListener.revoked); assertEquals(2, rebalanceListener.assignedCount); assertEquals(newAssignmentSet, rebalanceListener.assigned); } @@ -401,17 +450,17 @@ public class ConsumerCoordinatorTest { public void testWakeupDuringJoin() { final String consumerId = "leader"; - subscriptions.subscribe(singleton(topicName), rebalanceListener); + subscriptions.subscribe(singleton(topic1), rebalanceListener); // ensure metadata is up-to-date for leader - metadata.setTopics(singletonList(topicName)); + metadata.setTopics(singletonList(topic1)); metadata.update(cluster, time.milliseconds()); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); - Map<String, List<String>> memberSubscriptions = Collections.singletonMap(consumerId, singletonList(topicName)); - partitionAssignor.prepare(Collections.singletonMap(consumerId, singletonList(tp))); + Map<String, List<String>> memberSubscriptions = Collections.singletonMap(consumerId, singletonList(topic1)); + partitionAssignor.prepare(Collections.singletonMap(consumerId, singletonList(t1p))); // prepare only the first half of the join and then trigger the wakeup client.prepareResponse(joinGroupLeaderResponse(1, consumerId, memberSubscriptions, Errors.NONE.code())); @@ -424,22 +473,22 @@ public class ConsumerCoordinatorTest { } // now complete the second half - client.prepareResponse(syncGroupResponse(singletonList(tp), Errors.NONE.code())); + client.prepareResponse(syncGroupResponse(singletonList(t1p), Errors.NONE.code())); coordinator.poll(time.milliseconds()); assertFalse(coordinator.needRejoin()); - assertEquals(singleton(tp), subscriptions.assignedPartitions()); + assertEquals(singleton(t1p), subscriptions.assignedPartitions()); assertEquals(1, rebalanceListener.revokedCount); assertEquals(Collections.emptySet(), rebalanceListener.revoked); assertEquals(1, rebalanceListener.assignedCount); - assertEquals(singleton(tp), rebalanceListener.assigned); + assertEquals(singleton(t1p), rebalanceListener.assigned); } @Test public void testNormalJoinGroupFollower() { final String consumerId = "consumer"; - subscriptions.subscribe(singleton(topicName), rebalanceListener); + subscriptions.subscribe(singleton(topic1), rebalanceListener); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); @@ -454,28 +503,68 @@ public class ConsumerCoordinatorTest { sync.generationId() == 1 && sync.groupAssignment().isEmpty(); } - }, syncGroupResponse(singletonList(tp), Errors.NONE.code())); + }, syncGroupResponse(singletonList(t1p), Errors.NONE.code())); coordinator.joinGroupIfNeeded(); assertFalse(coordinator.needRejoin()); - assertEquals(singleton(tp), subscriptions.assignedPartitions()); + assertEquals(singleton(t1p), subscriptions.assignedPartitions()); + assertEquals(singleton(topic1), subscriptions.groupSubscription()); assertEquals(1, rebalanceListener.revokedCount); + assertEquals(Collections.emptySet(), rebalanceListener.revoked); assertEquals(1, rebalanceListener.assignedCount); - assertEquals(singleton(tp), rebalanceListener.assigned); + assertEquals(singleton(t1p), rebalanceListener.assigned); + } + + @Test + public void testPatternJoinGroupFollower() { + final String consumerId = "consumer"; + + subscriptions.subscribe(Pattern.compile("test.*"), rebalanceListener); + + // partially update the metadata with one topic first, + // let the leader to refresh metadata during assignment + metadata.setTopics(singletonList(topic1)); + metadata.update(TestUtils.singletonCluster(topic1, 1), time.milliseconds()); + + client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); + coordinator.ensureCoordinatorReady(); + + // normal join group + client.prepareResponse(joinGroupFollowerResponse(1, consumerId, "leader", Errors.NONE.code())); + client.prepareResponse(new MockClient.RequestMatcher() { + @Override + public boolean matches(AbstractRequest body) { + SyncGroupRequest sync = (SyncGroupRequest) body; + return sync.memberId().equals(consumerId) && + sync.generationId() == 1 && + sync.groupAssignment().isEmpty(); + } + }, syncGroupResponse(Arrays.asList(t1p, t2p), Errors.NONE.code())); + // expect client to force updating the metadata, if yes gives it both topics + client.prepareMetadataUpdate(cluster); + + coordinator.joinGroupIfNeeded(); + + assertFalse(coordinator.needRejoin()); + assertEquals(2, subscriptions.assignedPartitions().size()); + assertEquals(2, subscriptions.subscription().size()); + assertEquals(1, rebalanceListener.revokedCount); + assertEquals(1, rebalanceListener.assignedCount); + assertEquals(2, rebalanceListener.assigned.size()); } @Test public void testLeaveGroupOnClose() { final String consumerId = "consumer"; - subscriptions.subscribe(singleton(topicName), rebalanceListener); + subscriptions.subscribe(singleton(topic1), rebalanceListener); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); client.prepareResponse(joinGroupFollowerResponse(1, consumerId, "leader", Errors.NONE.code())); - client.prepareResponse(syncGroupResponse(singletonList(tp), Errors.NONE.code())); + client.prepareResponse(syncGroupResponse(singletonList(t1p), Errors.NONE.code())); coordinator.joinGroupIfNeeded(); final AtomicBoolean received = new AtomicBoolean(false); @@ -496,13 +585,13 @@ public class ConsumerCoordinatorTest { public void testMaybeLeaveGroup() { final String consumerId = "consumer"; - subscriptions.subscribe(singleton(topicName), rebalanceListener); + subscriptions.subscribe(singleton(topic1), rebalanceListener); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); client.prepareResponse(joinGroupFollowerResponse(1, consumerId, "leader", Errors.NONE.code())); - client.prepareResponse(syncGroupResponse(singletonList(tp), Errors.NONE.code())); + client.prepareResponse(syncGroupResponse(singletonList(t1p), Errors.NONE.code())); coordinator.joinGroupIfNeeded(); final AtomicBoolean received = new AtomicBoolean(false); @@ -526,7 +615,7 @@ public class ConsumerCoordinatorTest { public void testUnexpectedErrorOnSyncGroup() { final String consumerId = "consumer"; - subscriptions.subscribe(singleton(topicName), rebalanceListener); + subscriptions.subscribe(singleton(topic1), rebalanceListener); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); @@ -541,7 +630,7 @@ public class ConsumerCoordinatorTest { public void testUnknownMemberIdOnSyncGroup() { final String consumerId = "consumer"; - subscriptions.subscribe(singleton(topicName), rebalanceListener); + subscriptions.subscribe(singleton(topic1), rebalanceListener); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); @@ -558,19 +647,19 @@ public class ConsumerCoordinatorTest { return joinRequest.memberId().equals(JoinGroupRequest.UNKNOWN_MEMBER_ID); } }, joinGroupFollowerResponse(2, consumerId, "leader", Errors.NONE.code())); - client.prepareResponse(syncGroupResponse(singletonList(tp), Errors.NONE.code())); + client.prepareResponse(syncGroupResponse(singletonList(t1p), Errors.NONE.code())); coordinator.joinGroupIfNeeded(); assertFalse(coordinator.needRejoin()); - assertEquals(singleton(tp), subscriptions.assignedPartitions()); + assertEquals(singleton(t1p), subscriptions.assignedPartitions()); } @Test public void testRebalanceInProgressOnSyncGroup() { final String consumerId = "consumer"; - subscriptions.subscribe(singleton(topicName), rebalanceListener); + subscriptions.subscribe(singleton(topic1), rebalanceListener); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); @@ -581,19 +670,19 @@ public class ConsumerCoordinatorTest { // then let the full join/sync finish successfully client.prepareResponse(joinGroupFollowerResponse(2, consumerId, "leader", Errors.NONE.code())); - client.prepareResponse(syncGroupResponse(singletonList(tp), Errors.NONE.code())); + client.prepareResponse(syncGroupResponse(singletonList(t1p), Errors.NONE.code())); coordinator.joinGroupIfNeeded(); assertFalse(coordinator.needRejoin()); - assertEquals(singleton(tp), subscriptions.assignedPartitions()); + assertEquals(singleton(t1p), subscriptions.assignedPartitions()); } @Test public void testIllegalGenerationOnSyncGroup() { final String consumerId = "consumer"; - subscriptions.subscribe(singleton(topicName), rebalanceListener); + subscriptions.subscribe(singleton(topic1), rebalanceListener); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); @@ -610,12 +699,12 @@ public class ConsumerCoordinatorTest { return joinRequest.memberId().equals(JoinGroupRequest.UNKNOWN_MEMBER_ID); } }, joinGroupFollowerResponse(2, consumerId, "leader", Errors.NONE.code())); - client.prepareResponse(syncGroupResponse(singletonList(tp), Errors.NONE.code())); + client.prepareResponse(syncGroupResponse(singletonList(t1p), Errors.NONE.code())); coordinator.joinGroupIfNeeded(); assertFalse(coordinator.needRejoin()); - assertEquals(singleton(tp), subscriptions.assignedPartitions()); + assertEquals(singleton(t1p), subscriptions.assignedPartitions()); } @Test @@ -623,27 +712,27 @@ public class ConsumerCoordinatorTest { final String consumerId = "consumer"; // ensure metadata is up-to-date for leader - metadata.setTopics(singletonList(topicName)); + metadata.setTopics(singletonList(topic1)); metadata.update(cluster, time.milliseconds()); - subscriptions.subscribe(singleton(topicName), rebalanceListener); + subscriptions.subscribe(singleton(topic1), rebalanceListener); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); - Map<String, List<String>> memberSubscriptions = Collections.singletonMap(consumerId, singletonList(topicName)); - partitionAssignor.prepare(Collections.singletonMap(consumerId, singletonList(tp))); + Map<String, List<String>> memberSubscriptions = Collections.singletonMap(consumerId, singletonList(topic1)); + partitionAssignor.prepare(Collections.singletonMap(consumerId, singletonList(t1p))); // the leader is responsible for picking up metadata changes and forcing a group rebalance client.prepareResponse(joinGroupLeaderResponse(1, consumerId, memberSubscriptions, Errors.NONE.code())); - client.prepareResponse(syncGroupResponse(singletonList(tp), Errors.NONE.code())); + client.prepareResponse(syncGroupResponse(singletonList(t1p), Errors.NONE.code())); coordinator.poll(time.milliseconds()); assertFalse(coordinator.needRejoin()); // a new partition is added to the topic - metadata.update(TestUtils.singletonCluster(topicName, 2), time.milliseconds()); + metadata.update(TestUtils.singletonCluster(topic1, 2), time.milliseconds()); // we should detect the change and ask for reassignment assertTrue(coordinator.needRejoin()); @@ -670,7 +759,7 @@ public class ConsumerCoordinatorTest { // prepare initial rebalance Map<String, List<String>> memberSubscriptions = Collections.singletonMap(consumerId, topics); - partitionAssignor.prepare(Collections.singletonMap(consumerId, Arrays.asList(tp1))); + partitionAssignor.prepare(Collections.singletonMap(consumerId, Collections.singletonList(tp1))); client.prepareResponse(joinGroupLeaderResponse(1, consumerId, memberSubscriptions, Errors.NONE.code())); client.prepareResponse(new MockClient.RequestMatcher() { @@ -689,7 +778,7 @@ public class ConsumerCoordinatorTest { } return false; } - }, syncGroupResponse(Arrays.asList(tp1), Errors.NONE.code())); + }, syncGroupResponse(Collections.singletonList(tp1), Errors.NONE.code())); // the metadata update should trigger a second rebalance client.prepareResponse(joinGroupLeaderResponse(2, consumerId, memberSubscriptions, Errors.NONE.code())); @@ -725,36 +814,36 @@ public class ConsumerCoordinatorTest { public void testRejoinGroup() { String otherTopic = "otherTopic"; - subscriptions.subscribe(singleton(topicName), rebalanceListener); + subscriptions.subscribe(singleton(topic1), rebalanceListener); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); // join the group once client.prepareResponse(joinGroupFollowerResponse(1, "consumer", "leader", Errors.NONE.code())); - client.prepareResponse(syncGroupResponse(singletonList(tp), Errors.NONE.code())); + client.prepareResponse(syncGroupResponse(singletonList(t1p), Errors.NONE.code())); coordinator.joinGroupIfNeeded(); assertEquals(1, rebalanceListener.revokedCount); assertTrue(rebalanceListener.revoked.isEmpty()); assertEquals(1, rebalanceListener.assignedCount); - assertEquals(singleton(tp), rebalanceListener.assigned); + assertEquals(singleton(t1p), rebalanceListener.assigned); // and join the group again - subscriptions.subscribe(new HashSet<>(Arrays.asList(topicName, otherTopic)), rebalanceListener); + subscriptions.subscribe(new HashSet<>(Arrays.asList(topic1, otherTopic)), rebalanceListener); client.prepareResponse(joinGroupFollowerResponse(2, "consumer", "leader", Errors.NONE.code())); - client.prepareResponse(syncGroupResponse(singletonList(tp), Errors.NONE.code())); + client.prepareResponse(syncGroupResponse(singletonList(t1p), Errors.NONE.code())); coordinator.joinGroupIfNeeded(); assertEquals(2, rebalanceListener.revokedCount); - assertEquals(singleton(tp), rebalanceListener.revoked); + assertEquals(singleton(t1p), rebalanceListener.revoked); assertEquals(2, rebalanceListener.assignedCount); - assertEquals(singleton(tp), rebalanceListener.assigned); + assertEquals(singleton(t1p), rebalanceListener.assigned); } @Test public void testDisconnectInJoin() { - subscriptions.subscribe(singleton(topicName), rebalanceListener); + subscriptions.subscribe(singleton(topic1), rebalanceListener); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); @@ -763,19 +852,19 @@ public class ConsumerCoordinatorTest { client.prepareResponse(joinGroupFollowerResponse(1, "consumer", "leader", Errors.NONE.code()), true); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); client.prepareResponse(joinGroupFollowerResponse(1, "consumer", "leader", Errors.NONE.code())); - client.prepareResponse(syncGroupResponse(singletonList(tp), Errors.NONE.code())); + client.prepareResponse(syncGroupResponse(singletonList(t1p), Errors.NONE.code())); coordinator.joinGroupIfNeeded(); assertFalse(coordinator.needRejoin()); - assertEquals(singleton(tp), subscriptions.assignedPartitions()); + assertEquals(singleton(t1p), subscriptions.assignedPartitions()); assertEquals(1, rebalanceListener.revokedCount); assertEquals(1, rebalanceListener.assignedCount); - assertEquals(singleton(tp), rebalanceListener.assigned); + assertEquals(singleton(t1p), rebalanceListener.assigned); } @Test(expected = ApiException.class) public void testInvalidSessionTimeout() { - subscriptions.subscribe(singleton(topicName), rebalanceListener); + subscriptions.subscribe(singleton(topic1), rebalanceListener); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); @@ -787,19 +876,19 @@ public class ConsumerCoordinatorTest { @Test public void testCommitOffsetOnly() { - subscriptions.assignFromUser(singleton(tp)); + subscriptions.assignFromUser(singleton(t1p)); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); - client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.NONE.code()))); + client.prepareResponse(offsetCommitResponse(Collections.singletonMap(t1p, Errors.NONE.code()))); AtomicBoolean success = new AtomicBoolean(false); - coordinator.commitOffsetsAsync(Collections.singletonMap(tp, new OffsetAndMetadata(100L)), callback(success)); + coordinator.commitOffsetsAsync(Collections.singletonMap(t1p, new OffsetAndMetadata(100L)), callback(success)); coordinator.invokeCompletedOffsetCommitCallbacks(); assertTrue(success.get()); - assertEquals(100L, subscriptions.committed(tp).offset()); + assertEquals(100L, subscriptions.committed(t1p).offset()); } @Test @@ -809,22 +898,22 @@ public class ConsumerCoordinatorTest { ConsumerCoordinator coordinator = buildCoordinator(new Metrics(), assignors, ConsumerConfig.DEFAULT_EXCLUDE_INTERNAL_TOPICS, true); - subscriptions.subscribe(singleton(topicName), rebalanceListener); + subscriptions.subscribe(singleton(topic1), rebalanceListener); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); client.prepareResponse(joinGroupFollowerResponse(1, consumerId, "leader", Errors.NONE.code())); - client.prepareResponse(syncGroupResponse(singletonList(tp), Errors.NONE.code())); + client.prepareResponse(syncGroupResponse(singletonList(t1p), Errors.NONE.code())); coordinator.joinGroupIfNeeded(); - subscriptions.seek(tp, 100); + subscriptions.seek(t1p, 100); - client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.NONE.code()))); + client.prepareResponse(offsetCommitResponse(Collections.singletonMap(t1p, Errors.NONE.code()))); time.sleep(autoCommitIntervalMs); coordinator.poll(time.milliseconds()); - assertEquals(100L, subscriptions.committed(tp).offset()); + assertEquals(100L, subscriptions.committed(t1p).offset()); } @Test @@ -834,7 +923,7 @@ public class ConsumerCoordinatorTest { ConsumerCoordinator coordinator = buildCoordinator(new Metrics(), assignors, ConsumerConfig.DEFAULT_EXCLUDE_INTERNAL_TOPICS, true); - subscriptions.subscribe(singleton(topicName), rebalanceListener); + subscriptions.subscribe(singleton(topic1), rebalanceListener); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); @@ -844,16 +933,16 @@ public class ConsumerCoordinatorTest { consumerClient.poll(0); client.prepareResponse(joinGroupFollowerResponse(1, consumerId, "leader", Errors.NONE.code())); - client.prepareResponse(syncGroupResponse(singletonList(tp), Errors.NONE.code())); + client.prepareResponse(syncGroupResponse(singletonList(t1p), Errors.NONE.code())); coordinator.joinGroupIfNeeded(); - subscriptions.seek(tp, 100); + subscriptions.seek(t1p, 100); - client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.NONE.code()))); + client.prepareResponse(offsetCommitResponse(Collections.singletonMap(t1p, Errors.NONE.code()))); time.sleep(autoCommitIntervalMs); coordinator.poll(time.milliseconds()); - assertEquals(100L, subscriptions.committed(tp).offset()); + assertEquals(100L, subscriptions.committed(t1p).offset()); } @Test @@ -861,17 +950,17 @@ public class ConsumerCoordinatorTest { ConsumerCoordinator coordinator = buildCoordinator(new Metrics(), assignors, ConsumerConfig.DEFAULT_EXCLUDE_INTERNAL_TOPICS, true); - subscriptions.assignFromUser(singleton(tp)); - subscriptions.seek(tp, 100); + subscriptions.assignFromUser(singleton(t1p)); + subscriptions.seek(t1p, 100); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); - client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.NONE.code()))); + client.prepareResponse(offsetCommitResponse(Collections.singletonMap(t1p, Errors.NONE.code()))); time.sleep(autoCommitIntervalMs); coordinator.poll(time.milliseconds()); - assertEquals(100L, subscriptions.committed(tp).offset()); + assertEquals(100L, subscriptions.committed(t1p).offset()); } @Test @@ -879,15 +968,15 @@ public class ConsumerCoordinatorTest { ConsumerCoordinator coordinator = buildCoordinator(new Metrics(), assignors, ConsumerConfig.DEFAULT_EXCLUDE_INTERNAL_TOPICS, true); - subscriptions.assignFromUser(singleton(tp)); - subscriptions.seek(tp, 100); + subscriptions.assignFromUser(singleton(t1p)); + subscriptions.seek(t1p, 100); // no commit initially since coordinator is unknown consumerClient.poll(0); time.sleep(autoCommitIntervalMs); consumerClient.poll(0); - assertNull(subscriptions.committed(tp)); + assertNull(subscriptions.committed(t1p)); // now find the coordinator client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); @@ -895,28 +984,28 @@ public class ConsumerCoordinatorTest { // sleep only for the retry backoff time.sleep(retryBackoffMs); - client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.NONE.code()))); + client.prepareResponse(offsetCommitResponse(Collections.singletonMap(t1p, Errors.NONE.code()))); coordinator.poll(time.milliseconds()); - assertEquals(100L, subscriptions.committed(tp).offset()); + assertEquals(100L, subscriptions.committed(t1p).offset()); } @Test public void testCommitOffsetMetadata() { - subscriptions.assignFromUser(singleton(tp)); + subscriptions.assignFromUser(singleton(t1p)); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); - client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.NONE.code()))); + client.prepareResponse(offsetCommitResponse(Collections.singletonMap(t1p, Errors.NONE.code()))); AtomicBoolean success = new AtomicBoolean(false); - coordinator.commitOffsetsAsync(Collections.singletonMap(tp, new OffsetAndMetadata(100L, "hello")), callback(success)); + coordinator.commitOffsetsAsync(Collections.singletonMap(t1p, new OffsetAndMetadata(100L, "hello")), callback(success)); coordinator.invokeCompletedOffsetCommitCallbacks(); assertTrue(success.get()); - assertEquals(100L, subscriptions.committed(tp).offset()); - assertEquals("hello", subscriptions.committed(tp).metadata()); + assertEquals(100L, subscriptions.committed(t1p).offset()); + assertEquals("hello", subscriptions.committed(t1p).metadata()); } @Test @@ -924,8 +1013,8 @@ public class ConsumerCoordinatorTest { int invokedBeforeTest = defaultOffsetCommitCallback.invoked; client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); - client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.NONE.code()))); - coordinator.commitOffsetsAsync(Collections.singletonMap(tp, new OffsetAndMetadata(100L)), null); + client.prepareResponse(offsetCommitResponse(Collections.singletonMap(t1p, Errors.NONE.code()))); + coordinator.commitOffsetsAsync(Collections.singletonMap(t1p, new OffsetAndMetadata(100L)), null); coordinator.invokeCompletedOffsetCommitCallbacks(); assertEquals(invokedBeforeTest + 1, defaultOffsetCommitCallback.invoked); assertNull(defaultOffsetCommitCallback.exception); @@ -934,20 +1023,23 @@ public class ConsumerCoordinatorTest { @Test public void testCommitAfterLeaveGroup() { // enable auto-assignment - subscriptions.subscribe(singleton(topicName), rebalanceListener); + subscriptions.subscribe(singleton(topic1), rebalanceListener); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); client.prepareResponse(joinGroupFollowerResponse(1, "consumer", "leader", Errors.NONE.code())); - client.prepareResponse(syncGroupResponse(singletonList(tp), Errors.NONE.code())); + client.prepareResponse(syncGroupResponse(singletonList(t1p), Errors.NONE.code())); + + client.prepareMetadataUpdate(cluster); + coordinator.joinGroupIfNeeded(); // now switch to manual assignment client.prepareResponse(new LeaveGroupResponse(Errors.NONE.code())); subscriptions.unsubscribe(); coordinator.maybeLeaveGroup(); - subscriptions.assignFromUser(singleton(tp)); + subscriptions.assignFromUser(singleton(t1p)); // the client should not reuse generation/memberId from auto-subscribed generation client.prepareResponse(new MockClient.RequestMatcher() { @@ -957,10 +1049,10 @@ public class ConsumerCoordinatorTest { return commitRequest.memberId().equals(OffsetCommitRequest.DEFAULT_MEMBER_ID) && commitRequest.generationId() == OffsetCommitRequest.DEFAULT_GENERATION_ID; } - }, offsetCommitResponse(Collections.singletonMap(tp, Errors.NONE.code()))); + }, offsetCommitResponse(Collections.singletonMap(t1p, Errors.NONE.code()))); AtomicBoolean success = new AtomicBoolean(false); - coordinator.commitOffsetsAsync(Collections.singletonMap(tp, new OffsetAndMetadata(100L)), callback(success)); + coordinator.commitOffsetsAsync(Collections.singletonMap(t1p, new OffsetAndMetadata(100L)), callback(success)); coordinator.invokeCompletedOffsetCommitCallbacks(); assertTrue(success.get()); } @@ -970,8 +1062,8 @@ public class ConsumerCoordinatorTest { int invokedBeforeTest = defaultOffsetCommitCallback.invoked; client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); - client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.GROUP_COORDINATOR_NOT_AVAILABLE.code()))); - coordinator.commitOffsetsAsync(Collections.singletonMap(tp, new OffsetAndMetadata(100L)), null); + client.prepareResponse(offsetCommitResponse(Collections.singletonMap(t1p, Errors.GROUP_COORDINATOR_NOT_AVAILABLE.code()))); + coordinator.commitOffsetsAsync(Collections.singletonMap(t1p, new OffsetAndMetadata(100L)), null); coordinator.invokeCompletedOffsetCommitCallbacks(); assertEquals(invokedBeforeTest + 1, defaultOffsetCommitCallback.invoked); assertTrue(defaultOffsetCommitCallback.exception instanceof RetriableCommitFailedException); @@ -984,8 +1076,8 @@ public class ConsumerCoordinatorTest { // async commit with coordinator not available MockCommitCallback cb = new MockCommitCallback(); - client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.GROUP_COORDINATOR_NOT_AVAILABLE.code()))); - coordinator.commitOffsetsAsync(Collections.singletonMap(tp, new OffsetAndMetadata(100L)), cb); + client.prepareResponse(offsetCommitResponse(Collections.singletonMap(t1p, Errors.GROUP_COORDINATOR_NOT_AVAILABLE.code()))); + coordinator.commitOffsetsAsync(Collections.singletonMap(t1p, new OffsetAndMetadata(100L)), cb); coordinator.invokeCompletedOffsetCommitCallbacks(); assertTrue(coordinator.coordinatorUnknown()); @@ -1000,8 +1092,8 @@ public class ConsumerCoordinatorTest { // async commit with not coordinator MockCommitCallback cb = new MockCommitCallback(); - client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.NOT_COORDINATOR_FOR_GROUP.code()))); - coordinator.commitOffsetsAsync(Collections.singletonMap(tp, new OffsetAndMetadata(100L)), cb); + client.prepareResponse(offsetCommitResponse(Collections.singletonMap(t1p, Errors.NOT_COORDINATOR_FOR_GROUP.code()))); + coordinator.commitOffsetsAsync(Collections.singletonMap(t1p, new OffsetAndMetadata(100L)), cb); coordinator.invokeCompletedOffsetCommitCallbacks(); assertTrue(coordinator.coordinatorUnknown()); @@ -1016,8 +1108,8 @@ public class ConsumerCoordinatorTest { // async commit with coordinator disconnected MockCommitCallback cb = new MockCommitCallback(); - client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.NONE.code())), true); - coordinator.commitOffsetsAsync(Collections.singletonMap(tp, new OffsetAndMetadata(100L)), cb); + client.prepareResponse(offsetCommitResponse(Collections.singletonMap(t1p, Errors.NONE.code())), true); + coordinator.commitOffsetsAsync(Collections.singletonMap(t1p, new OffsetAndMetadata(100L)), cb); coordinator.invokeCompletedOffsetCommitCallbacks(); assertTrue(coordinator.coordinatorUnknown()); @@ -1031,10 +1123,10 @@ public class ConsumerCoordinatorTest { coordinator.ensureCoordinatorReady(); // sync commit with coordinator disconnected (should connect, get metadata, and then submit the commit request) - client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.NOT_COORDINATOR_FOR_GROUP.code()))); + client.prepareResponse(offsetCommitResponse(Collections.singletonMap(t1p, Errors.NOT_COORDINATOR_FOR_GROUP.code()))); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); - client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.NONE.code()))); - coordinator.commitOffsetsSync(Collections.singletonMap(tp, new OffsetAndMetadata(100L)), Long.MAX_VALUE); + client.prepareResponse(offsetCommitResponse(Collections.singletonMap(t1p, Errors.NONE.code()))); + coordinator.commitOffsetsSync(Collections.singletonMap(t1p, new OffsetAndMetadata(100L)), Long.MAX_VALUE); } @Test @@ -1043,10 +1135,10 @@ public class ConsumerCoordinatorTest { coordinator.ensureCoordinatorReady(); // sync commit with coordinator disconnected (should connect, get metadata, and then submit the commit request) - client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.GROUP_COORDINATOR_NOT_AVAILABLE.code()))); + client.prepareResponse(offsetCommitResponse(Collections.singletonMap(t1p, Errors.GROUP_COORDINATOR_NOT_AVAILABLE.code()))); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); - client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.NONE.code()))); - coordinator.commitOffsetsSync(Collections.singletonMap(tp, new OffsetAndMetadata(100L)), Long.MAX_VALUE); + client.prepareResponse(offsetCommitResponse(Collections.singletonMap(t1p, Errors.NONE.code()))); + coordinator.commitOffsetsSync(Collections.singletonMap(t1p, new OffsetAndMetadata(100L)), Long.MAX_VALUE); } @Test @@ -1055,10 +1147,10 @@ public class ConsumerCoordinatorTest { coordinator.ensureCoordinatorReady(); // sync commit with coordinator disconnected (should connect, get metadata, and then submit the commit request) - client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.NONE.code())), true); + client.prepareResponse(offsetCommitResponse(Collections.singletonMap(t1p, Errors.NONE.code())), true); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); - client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.NONE.code()))); - coordinator.commitOffsetsSync(Collections.singletonMap(tp, new OffsetAndMetadata(100L)), Long.MAX_VALUE); + client.prepareResponse(offsetCommitResponse(Collections.singletonMap(t1p, Errors.NONE.code()))); + coordinator.commitOffsetsSync(Collections.singletonMap(t1p, new OffsetAndMetadata(100L)), Long.MAX_VALUE); } @Test(expected = KafkaException.class) @@ -1066,8 +1158,8 @@ public class ConsumerCoordinatorTest { client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); - client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.UNKNOWN_TOPIC_OR_PARTITION.code()))); - coordinator.commitOffsetsSync(Collections.singletonMap(tp, new OffsetAndMetadata(100L, "metadata")), Long.MAX_VALUE); + client.prepareResponse(offsetCommitResponse(Collections.singletonMap(t1p, Errors.UNKNOWN_TOPIC_OR_PARTITION.code()))); + coordinator.commitOffsetsSync(Collections.singletonMap(t1p, new OffsetAndMetadata(100L, "metadata")), Long.MAX_VALUE); } @Test(expected = OffsetMetadataTooLarge.class) @@ -1076,8 +1168,8 @@ public class ConsumerCoordinatorTest { client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); - client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.OFFSET_METADATA_TOO_LARGE.code()))); - coordinator.commitOffsetsSync(Collections.singletonMap(tp, new OffsetAndMetadata(100L, "metadata")), Long.MAX_VALUE); + client.prepareResponse(offsetCommitResponse(Collections.singletonMap(t1p, Errors.OFFSET_METADATA_TOO_LARGE.code()))); + coordinator.commitOffsetsSync(Collections.singletonMap(t1p, new OffsetAndMetadata(100L, "metadata")), Long.MAX_VALUE); } @Test(expected = CommitFailedException.class) @@ -1086,8 +1178,8 @@ public class ConsumerCoordinatorTest { client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); - client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.ILLEGAL_GENERATION.code()))); - coordinator.commitOffsetsSync(Collections.singletonMap(tp, new OffsetAndMetadata(100L, "metadata")), Long.MAX_VALUE); + client.prepareResponse(offsetCommitResponse(Collections.singletonMap(t1p, Errors.ILLEGAL_GENERATION.code()))); + coordinator.commitOffsetsSync(Collections.singletonMap(t1p, new OffsetAndMetadata(100L, "metadata")), Long.MAX_VALUE); } @Test(expected = CommitFailedException.class) @@ -1096,8 +1188,8 @@ public class ConsumerCoordinatorTest { client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); - client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.UNKNOWN_MEMBER_ID.code()))); - coordinator.commitOffsetsSync(Collections.singletonMap(tp, new OffsetAndMetadata(100L, "metadata")), Long.MAX_VALUE); + client.prepareResponse(offsetCommitResponse(Collections.singletonMap(t1p, Errors.UNKNOWN_MEMBER_ID.code()))); + coordinator.commitOffsetsSync(Collections.singletonMap(t1p, new OffsetAndMetadata(100L, "metadata")), Long.MAX_VALUE); } @Test(expected = CommitFailedException.class) @@ -1106,8 +1198,8 @@ public class ConsumerCoordinatorTest { client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); - client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.REBALANCE_IN_PROGRESS.code()))); - coordinator.commitOffsetsSync(Collections.singletonMap(tp, new OffsetAndMetadata(100L, "metadata")), Long.MAX_VALUE); + client.prepareResponse(offsetCommitResponse(Collections.singletonMap(t1p, Errors.REBALANCE_IN_PROGRESS.code()))); + coordinator.commitOffsetsSync(Collections.singletonMap(t1p, new OffsetAndMetadata(100L, "metadata")), Long.MAX_VALUE); } @Test(expected = KafkaException.class) @@ -1116,21 +1208,21 @@ public class ConsumerCoordinatorTest { coordinator.ensureCoordinatorReady(); // sync commit with invalid partitions should throw if we have no callback - client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp, Errors.UNKNOWN.code())), false); - coordinator.commitOffsetsSync(Collections.singletonMap(tp, new OffsetAndMetadata(100L)), Long.MAX_VALUE); + client.prepareResponse(offsetCommitResponse(Collections.singletonMap(t1p, Errors.UNKNOWN.code())), false); + coordinator.commitOffsetsSync(Collections.singletonMap(t1p, new OffsetAndMetadata(100L)), Long.MAX_VALUE); } @Test(expected = IllegalArgumentException.class) public void testCommitSyncNegativeOffset() { client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); - coordinator.commitOffsetsSync(Collections.singletonMap(tp, new OffsetAndMetadata(-1L)), Long.MAX_VALUE); + coordinator.commitOffsetsSync(Collections.singletonMap(t1p, new OffsetAndMetadata(-1L)), Long.MAX_VALUE); } @Test public void testCommitAsyncNegativeOffset() { int invokedBeforeTest = defaultOffsetCommitCallback.invoked; client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); - coordinator.commitOffsetsAsync(Collections.singletonMap(tp, new OffsetAndMetadata(-1L)), null); + coordinator.commitOffsetsAsync(Collections.singletonMap(t1p, new OffsetAndMetadata(-1L)), null); coordinator.invokeCompletedOffsetCommitCallbacks(); assertEquals(invokedBeforeTest + 1, defaultOffsetCommitCallback.invoked); assertTrue(defaultOffsetCommitCallback.exception instanceof IllegalArgumentException); @@ -1141,12 +1233,12 @@ public class ConsumerCoordinatorTest { client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); - subscriptions.assignFromUser(singleton(tp)); + subscriptions.assignFromUser(singleton(t1p)); subscriptions.needRefreshCommits(); - client.prepareResponse(offsetFetchResponse(tp, Errors.NONE, "", 100L)); + client.prepareResponse(offsetFetchResponse(t1p, Errors.NONE, "", 100L)); coordinator.refreshCommittedOffsetsIfNeeded(); assertFalse(subscriptions.refreshCommitsNeeded()); - assertEquals(100L, subscriptions.committed(tp).offset()); + assertEquals(100L, subscriptions.committed(t1p).offset()); } @Test @@ -1154,13 +1246,13 @@ public class ConsumerCoordinatorTest { client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); - subscriptions.assignFromUser(singleton(tp)); + subscriptions.assignFromUser(singleton(t1p)); subscriptions.needRefreshCommits(); client.prepareResponse(offsetFetchResponse(Errors.GROUP_LOAD_IN_PROGRESS)); - client.prepareResponse(offsetFetchResponse(tp, Errors.NONE, "", 100L)); + client.prepareResponse(offsetFetchResponse(t1p, Errors.NONE, "", 100L)); coordinator.refreshCommittedOffsetsIfNeeded(); assertFalse(subscriptions.refreshCommitsNeeded()); - assertEquals(100L, subscriptions.committed(tp).offset()); + assertEquals(100L, subscriptions.committed(t1p).offset()); } @Test @@ -1168,7 +1260,7 @@ public class ConsumerCoordinatorTest { client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); - subscriptions.assignFromUser(singleton(tp)); + subscriptions.assignFromUser(singleton(t1p)); subscriptions.needRefreshCommits(); client.prepareResponse(offsetFetchResponse(Errors.GROUP_AUTHORIZATION_FAILED)); try { @@ -1184,9 +1276,9 @@ public class ConsumerCoordinatorTest { client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); - subscriptions.assignFromUser(singleton(tp)); + subscriptions.assignFromUser(singleton(t1p)); subscriptions.needRefreshCommits(); - client.prepareResponse(offsetFetchResponse(tp, Errors.UNKNOWN_TOPIC_OR_PARTITION, "", 100L)); + client.prepareResponse(offsetFetchResponse(t1p, Errors.UNKNOWN_TOPIC_OR_PARTITION, "", 100L)); coordinator.refreshCommittedOffsetsIfNeeded(); } @@ -1195,14 +1287,14 @@ public class ConsumerCoordinatorTest { client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); - subscriptions.assignFromUser(singleton(tp)); + subscriptions.assignFromUser(singleton(t1p)); subscriptions.needRefreshCommits(); client.prepareResponse(offsetFetchResponse(Errors.NOT_COORDINATOR_FOR_GROUP)); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); - client.prepareResponse(offsetFetchResponse(tp, Errors.NONE, "", 100L)); + client.prepareResponse(offsetFetchResponse(t1p, Errors.NONE, "", 100L)); coordinator.refreshCommittedOffsetsIfNeeded(); assertFalse(subscriptions.refreshCommitsNeeded()); - assertEquals(100L, subscriptions.committed(tp).offset()); + assertEquals(100L, subscriptions.committed(t1p).offset()); } @Test @@ -1210,12 +1302,12 @@ public class ConsumerCoordinatorTest { client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); - subscriptions.assignFromUser(singleton(tp)); + subscriptions.assignFromUser(singleton(t1p)); subscriptions.needRefreshCommits(); - client.prepareResponse(offsetFetchResponse(tp, Errors.NONE, "", -1L)); + client.prepareResponse(offsetFetchResponse(t1p, Errors.NONE, "", -1L)); coordinator.refreshCommittedOffsetsIfNeeded(); assertFalse(subscriptions.refreshCommitsNeeded()); - assertEquals(null, subscriptions.committed(tp)); + assertEquals(null, subscriptions.committed(t1p)); } @Test @@ -1341,14 +1433,14 @@ public class ConsumerCoordinatorTest { client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); coordinator.ensureCoordinatorReady(); if (useGroupManagement) { - subscriptions.subscribe(singleton(topicName), rebalanceListener); + subscriptions.subscribe(singleton(topic1), rebalanceListener); client.prepareResponse(joinGroupFollowerResponse(1, consumerId, "leader", Errors.NONE.code())); - client.prepareResponse(syncGroupResponse(singletonList(tp), Errors.NONE.code())); + client.prepareResponse(syncGroupResponse(singletonList(t1p), Errors.NONE.code())); coordinator.joinGroupIfNeeded(); } else - subscriptions.assignFromUser(singleton(tp)); + subscriptions.assignFromUser(singleton(t1p)); - subscriptions.seek(tp, 100); + subscriptions.seek(t1p, 100); coordinator.poll(time.milliseconds()); return coordinator; http://git-wip-us.apache.org/repos/asf/kafka/blob/3400d0c3/clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java ---------------------------------------------------------------------- diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java index ef75754..55bf2a3 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java @@ -29,7 +29,6 @@ import java.util.Collections; import java.util.HashSet; import java.util.regex.Pattern; -import static java.util.Arrays.asList; import static java.util.Collections.singleton; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; @@ -77,7 +76,7 @@ public class SubscriptionStateTest { // assigned partitions should remain unchanged assertTrue(state.assignedPartitions().isEmpty()); - state.assignFromSubscribed(asList(t1p0)); + state.assignFromSubscribed(Collections.singletonList(t1p0)); // assigned partitions should immediately change assertEquals(singleton(t1p0), state.assignedPartitions()); @@ -96,21 +95,32 @@ public class SubscriptionStateTest { // assigned partitions should remain unchanged assertTrue(state.assignedPartitions().isEmpty()); - state.subscribeFromPattern(new HashSet<>(Arrays.asList(topic, topic1))); + state.subscribeFromPattern(new HashSet<>(Collections.singletonList(topic))); // assigned partitions should remain unchanged assertTrue(state.assignedPartitions().isEmpty()); - state.assignFromSubscribed(asList(tp1)); + state.assignFromSubscribed(Collections.singletonList(tp1)); // assigned partitions should immediately change assertEquals(singleton(tp1), state.assignedPartitions()); + assertEquals(singleton(topic), state.subscription()); + + state.assignFromSubscribed(Collections.singletonList(t1p0)); + // assigned partitions should immediately change + assertEquals(singleton(t1p0), state.assignedPartitions()); + assertEquals(singleton(topic), state.subscription()); state.subscribe(Pattern.compile(".*t"), rebalanceListener); // assigned partitions should remain unchanged - assertEquals(singleton(tp1), state.assignedPartitions()); + assertEquals(singleton(t1p0), state.assignedPartitions()); state.subscribeFromPattern(singleton(topic)); // assigned partitions should remain unchanged - assertEquals(singleton(tp1), state.assignedPartitions()); + assertEquals(singleton(t1p0), state.assignedPartitions()); + + state.assignFromSubscribed(Collections.singletonList(tp0)); + // assigned partitions should immediately change + assertEquals(singleton(tp0), state.assignedPartitions()); + assertEquals(singleton(topic), state.subscription()); state.unsubscribe(); // assigned partitions should immediately change @@ -139,11 +149,11 @@ public class SubscriptionStateTest { assertEquals(1, state.subscription().size()); assertTrue(state.assignedPartitions().isEmpty()); assertTrue(state.partitionsAutoAssigned()); - state.assignFromSubscribed(asList(tp0)); + state.assignFromSubscribed(Collections.singletonList(tp0)); state.seek(tp0, 1); state.committed(tp0, new OffsetAndMetadata(1)); assertAllPositions(tp0, 1L); - state.assignFromSubscribed(asList(tp1)); + state.assignFromSubscribed(Collections.singletonList(tp1)); assertTrue(state.isAssigned(tp1)); assertFalse(state.isAssigned(tp0)); assertFalse(state.isFetchable(tp1)); @@ -173,20 +183,28 @@ public class SubscriptionStateTest { @Test(expected = IllegalStateException.class) public void invalidPositionUpdate() { state.subscribe(singleton(topic), rebalanceListener); - state.assignFromSubscribed(asList(tp0)); + state.assignFromSubscribed(Collections.singletonList(tp0)); state.position(tp0, 0); } + @Test(expected = IllegalArgumentException.class) + public void cantAssignPartitionForUnsubscribedTopics() { + state.subscribe(singleton(topic), rebalanceListener); + state.assignFromSubscribed(Collections.singletonList(t1p0)); + } + + @Test(expected = IllegalArgumentException.class) + public void cantAssignPartitionForUnmatchedPattern() { + state.subscribe(Pattern.compile(".*t"), rebalanceListener); + state.subscribeFromPattern(new HashSet<>(Collections.singletonList(topic))); + state.assignFromSubscribed(Collections.singletonList(t1p0)); + } + @Test(expected = IllegalStateException.class) public void cantChangePositionForNonAssignedPartition() { state.position(tp0, 1); } - public void assertAllPositions(TopicPartition tp, Long offset) { - assertEquals(offset.longValue(), state.committed(tp).offset()); - assertEquals(offset, state.position(tp)); - } - @Test(expected = IllegalStateException.class) public void cantSubscribeTopicAndPattern() { state.subscribe(singleton(topic), rebalanceListener); @@ -240,7 +258,7 @@ public class SubscriptionStateTest { public void unsubscription() { state.subscribe(Pattern.compile(".*"), rebalanceListener); state.subscribeFromPattern(new HashSet<>(Arrays.asList(topic, topic1))); - state.assignFromSubscribed(asList(tp1)); + state.assignFromSubscribed(Collections.singletonList(tp1)); assertEquals(singleton(tp1), state.assignedPartitions()); state.unsubscribe(); @@ -255,6 +273,11 @@ public class SubscriptionStateTest { assertTrue(state.assignedPartitions().isEmpty()); } + private void assertAllPositions(TopicPartition tp, Long offset) { + assertEquals(offset.longValue(), state.committed(tp).offset()); + assertEquals(offset, state.position(tp)); + } + private static class MockRebalanceListener implements ConsumerRebalanceListener { public Collection<TopicPartition> revoked; public Collection<TopicPartition> assigned; http://git-wip-us.apache.org/repos/asf/kafka/blob/3400d0c3/streams/src/main/java/org/apache/kafka/streams/kstream/internals/AbstractStream.java ---------------------------------------------------------------------- diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/AbstractStream.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/AbstractStream.java index 31a3dc6..bcffce2 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/AbstractStream.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/AbstractStream.java @@ -37,12 +37,6 @@ public abstract class AbstractStream<K> { protected final String name; protected final Set<String> sourceNodes; - public AbstractStream(AbstractStream<K> stream) { - this.topology = stream.topology; - this.name = stream.name; - this.sourceNodes = stream.sourceNodes; - } - public AbstractStream(KStreamBuilder topology, String name, Set<String> sourceNodes) { if (sourceNodes == null || sourceNodes.isEmpty()) { throw new IllegalArgumentException("parameter <sourceNodes> must not be null or empty");
