mjsax commented on code in PR #22788:
URL: https://github.com/apache/kafka/pull/22788#discussion_r3626417837
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -19592,8 +19592,9 @@ public void
testStreamsGroupMemberJoiningWithStaleTopology() {
)
.build();
- assignor.prepareGroupAssignment(new
org.apache.kafka.coordinator.group.streams.assignor.GroupAssignment(Map.of(
- memberId,
org.apache.kafka.coordinator.group.streams.assignor.MemberAssignment.empty()
+ assignor.prepareGroupAssignment(new
org.apache.kafka.coordinator.group.api.streams.assignor.GroupAssignment(Map.of(
+ memberId,
(org.apache.kafka.coordinator.group.api.streams.assignor.MemberAssignment)
Review Comment:
`MemberAssignmentImpl implements MemberAssignment`, so this cast should not
be necessary?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/MockTaskAssignor.java:
##########
@@ -48,8 +49,8 @@ public void prepareGroupAssignment(Map<String, TasksTuple>
memberAssignments) {
Entry::getKey,
entry -> {
TasksTuple tasksTuple = entry.getValue();
- return new MemberAssignment(
- tasksTuple.activeTasks(),
tasksTuple.standbyTasks(), tasksTuple.warmupTasks());
+ return (MemberAssignment) new MemberAssignmentImpl(
Review Comment:
```suggestion
return new MemberAssignmentImpl(
```
##########
group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/MemberAssignmentState.java:
##########
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.api.streams.assignor;
+
+import org.apache.kafka.common.annotation.InterfaceAudience;
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Interface representing the current assignment state for a streams group
member, used by the
+ * {@link TaskAssignor} to compute a new, sticky target assignment.
Review Comment:
```suggestion
* {@link TaskAssignor} to compute a new target assignment.
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/MemberMetadataAndAssignmentImpl.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.streams.assignor;
+
+import
org.apache.kafka.coordinator.group.api.streams.assignor.MemberAssignmentMetadata;
+import
org.apache.kafka.coordinator.group.api.streams.assignor.MemberAssignmentState;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+/**
+ * Implementation of both the {@link MemberAssignmentMetadata} and the {@link
MemberAssignmentState}
+ * interfaces for a streams group member.
+ *
+ * @param instanceId The instance ID if provided.
+ * @param rackId The rack ID if provided.
+ * @param activeTasks Current active tasks.
+ * @param standbyTasks Current standby tasks.
+ * @param warmupTasks Current warm-up tasks.
+ * @param processId The process ID.
+ * @param clientTags The client tags for a rack-aware assignment.
+ * @param taskOffsets The last received cumulative task offsets of assigned
tasks or dormant tasks.
+ * @param taskEndOffsets The last received cumulative task end offsets of
assigned tasks or dormant tasks.
+ */
+public record MemberMetadataAndAssignmentImpl(Optional<String> instanceId,
+ Optional<String> rackId,
+ Map<String, Set<Integer>>
activeTasks,
+ Map<String, Set<Integer>>
standbyTasks,
+ Map<String, Set<Integer>>
warmupTasks,
+ String processId,
+ Map<String, String>
clientTags,
+ Map<String, Map<Integer,
Long>> taskOffsets,
+ Map<String, Map<Integer,
Long>> taskEndOffsets
+) implements MemberAssignmentMetadata, MemberAssignmentState {
+
+ public MemberMetadataAndAssignmentImpl {
+ Objects.requireNonNull(instanceId);
+ Objects.requireNonNull(rackId);
+ activeTasks =
Collections.unmodifiableMap(Objects.requireNonNull(activeTasks));
Review Comment:
It seems the fields are mixed weirdly -- should we re-order them, and first
pass in all 4 "metadata fields" followed by all "state fields" (applies to the
parameter list above and the constructor code.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/MemberAssignmentImpl.java:
##########
@@ -16,28 +16,31 @@
*/
package org.apache.kafka.coordinator.group.streams.assignor;
+import
org.apache.kafka.coordinator.group.api.streams.assignor.MemberAssignment;
+
+import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
/**
* The task assignment for a streams group member.
*
- * @param activeTasks The active tasks assigned to this member keyed by
subtopologyId.
- * @param standbyTasks The standby tasks assigned to this member keyed by
subtopologyId.
- * @param warmupTasks The warm-up tasks assigned to this member keyed by
subtopologyId.
+ * @param activeTasks The active tasks assigned to this member keyed by
subtopology Id.
+ * @param standbyTasks The standby tasks assigned to this member keyed by
subtopology Id.
+ * The maps are not made immutable, since the server-side
assignors rely on
+ * being able to mutate them while building new
assignments.
Review Comment:
Why is this sentence part of `@param standbyTasks` description? Does it only
apply to `standbyTasks` ? If it applied to both parameters, it should got above
the `@param` section.
Also, if the server really relies on mutability, it must be stated on
`MemberAssignment` _interface_ such that users who implement their own
task-assignors know about this requirement. Even if it makes me itchy that we
do require it -- it seems to be easy source for bug. Wondering if we should
drop this requirement and make a _deep copy_ instead inside the broker?
\cc @squah-confluent
##########
group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/MemberAssignmentState.java:
##########
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.api.streams.assignor;
+
+import org.apache.kafka.common.annotation.InterfaceAudience;
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Interface representing the current assignment state for a streams group
member, used by the
+ * {@link TaskAssignor} to compute a new, sticky target assignment.
+ *
+ * <p>The active and standby tasks are the member's current <em>target</em>
assignment (the last
+ * assignment committed for the member). Using the committed target as the
stickiness baseline keeps
+ * in-flight task moves converging rather than reverting them. Warm-up tasks
are not assigned by the
+ * assignor (they are decided during reconciliation); the tasks the member is
<em>currently</em>
+ * warming up are exposed here so that the assignor can take them into
account. As a result, a task
+ * that is being moved to this member can appear both as an active or standby
task (from the target)
+ * and as a warm-up task (currently in progress).
+ *
+ * <p>All accessors are keyed by subtopology ID. The task-set accessors
({@link #activeTasks()},
+ * {@link #standbyTasks()}, {@link #warmupTasks()}) map each subtopology ID to
its set of
+ * partitions, while {@link #taskOffsets()} and {@link #taskEndOffsets()} map
each subtopology ID
+ * to a map from partition to offset.
+ */
[email protected]
[email protected]
+public interface MemberAssignmentState {
+
+ /**
+ * @return The member's current target active tasks keyed by subtopology
Id.
Review Comment:
Is this really "target tasks"? Or rather "currently assigned active tasks" ?
Same question below.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -98,33 +108,33 @@ private void initialize(final GroupSpec groupSpec, final
TopologyDescriber topol
if (topologyDescriber.isStateful(subtopology))
localState.totalTasks += numberOfPartitions *
localState.numStandbyReplicas;
}
- localState.totalMembersWithActiveTaskCapacity =
groupSpec.members().size();
- localState.totalMembersWithTaskCapacity = groupSpec.members().size();
+ localState.totalMembersWithActiveTaskCapacity =
groupSpec.memberIds().size();
+ localState.totalMembersWithTaskCapacity = groupSpec.memberIds().size();
localState.activeTasksPerMember =
computeTasksPerMember(localState.totalActiveTasks,
localState.totalMembersWithActiveTaskCapacity);
localState.totalTasksPerMember =
computeTasksPerMember(localState.totalTasks,
localState.totalMembersWithTaskCapacity);
localState.processIdToState = new
HashMap<>(localState.totalMembersWithActiveTaskCapacity);
localState.activeTaskToPrevMember = new
HashMap<>(localState.totalActiveTasks);
localState.standbyTaskToPrevMember = new
HashMap<>(localState.numStandbyReplicas > 0 ? (localState.totalTasks -
localState.totalActiveTasks) / localState.numStandbyReplicas : 0);
- for (final Map.Entry<String, AssignmentMemberSpec> memberEntry :
groupSpec.members().entrySet()) {
- final String memberId = memberEntry.getKey();
- final String processId = memberEntry.getValue().processId();
+ for (final String memberId : groupSpec.memberIds()) {
+ final MemberAssignmentMetadata memberMetadata =
groupSpec.memberMetadata(memberId);
+ final MemberAssignmentState memberAssignmentState =
groupSpec.memberAssignmentState(memberId);
+ final String processId = memberMetadata.processId();
Review Comment:
```suggestion
final String processId =
groupSpec.memberMetadata(memberId).processId();
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java:
##########
@@ -234,22 +237,22 @@ public TargetAssignmentResult build() throws
TaskAssignorException {
}
newGroupAssignment = assignor.assign(
new GroupSpecImpl(
- Collections.unmodifiableMap(memberSpecs),
+ Collections.unmodifiableMap(memberMetadataMap),
assignmentConfigs
),
new TopologyMetadata(metadataImage,
topology.subtopologies().get())
);
} else {
newGroupAssignment = new GroupAssignment(
- memberSpecs.keySet().stream().collect(Collectors.toMap(x -> x,
x -> MemberAssignment.empty())));
+ memberMetadataMap.keySet().stream().collect(Collectors.toMap(x
-> x, x -> (MemberAssignment) MemberAssignmentImpl.empty())));
Review Comment:
`MemberAssignmentImpl implements MemberAssignment`, so this cast should not
be necessary?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilderTest.java:
##########
@@ -98,54 +100,80 @@ public void testCreateAssignmentMemberSpec(TaskRole
taskRole) {
.setInstanceId("instanceId")
.setProcessId("processId")
.setClientTags(clientTags)
+ .setAssignedTasks(TasksTupleWithEpochs.EMPTY)
.build();
TasksTuple assignment = mkTasksTuple(taskRole,
mkTasks(fooSubtopologyId, 1, 2, 3),
mkTasks(barSubtopologyId, 1, 2, 3)
);
- AssignmentMemberSpec assignmentMemberSpec = createAssignmentMemberSpec(
+ MemberMetadataAndAssignmentImpl memberMetadata =
createMemberMetadataAndAssignment(
member,
assignment,
MemberTaskOffsets.EMPTY
);
- assertEquals(new AssignmentMemberSpec(
+ assertEquals(new MemberMetadataAndAssignmentImpl(
Optional.of("instanceId"),
Optional.of("rackId"),
assignment.activeTasks(),
assignment.standbyTasks(),
- assignment.warmupTasks(),
+ Map.of(),
"processId",
clientTags,
Map.of(),
Map.of()
- ), assignmentMemberSpec);
+ ), memberMetadata);
}
@Test
- public void testCreateAssignmentMemberSpecPopulatesTaskOffsets() {
+ public void testCreateMemberMetadataAndAssignmentPopulatesTaskOffsets() {
String fooSubtopologyId = Uuid.randomUuid().toString();
StreamsGroupMember member = new StreamsGroupMember.Builder("member-id")
.setRackId("rackId")
.setInstanceId("instanceId")
.setProcessId("processId")
.setClientTags(Map.of())
+ .setAssignedTasks(TasksTupleWithEpochs.EMPTY)
Review Comment:
same question
##########
group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/GroupAssignment.java:
##########
@@ -24,6 +27,8 @@
*
* @param members The member assignments keyed by member ID.
*/
[email protected]
[email protected]
public record GroupAssignment(Map<String, MemberAssignment> members) {
Review Comment:
\cc @squah-confluent
##########
group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/MemberAssignmentState.java:
##########
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.api.streams.assignor;
+
+import org.apache.kafka.common.annotation.InterfaceAudience;
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Interface representing the current assignment state for a streams group
member, used by the
+ * {@link TaskAssignor} to compute a new, sticky target assignment.
Review Comment:
Guess c&p error from StickyAssignor :)
##########
group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/MemberAssignmentState.java:
##########
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.api.streams.assignor;
+
+import org.apache.kafka.common.annotation.InterfaceAudience;
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Interface representing the current assignment state for a streams group
member, used by the
+ * {@link TaskAssignor} to compute a new, sticky target assignment.
+ *
+ * <p>The active and standby tasks are the member's current <em>target</em>
assignment (the last
+ * assignment committed for the member). Using the committed target as the
stickiness baseline keeps
Review Comment:
More `StickyAssignor` left-overs. Need to be cleaned up.
This is a generic interface, which does not know anything about why
assignment strategy will be used.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/MockAssignor.java:
##########
@@ -56,13 +63,10 @@ public GroupAssignment assign(
}
// Copy existing assignment and fill temporary data structures
- for (Map.Entry<String, AssignmentMemberSpec> memberEntry :
groupSpec.members().entrySet()) {
- final String memberId = memberEntry.getKey();
- final AssignmentMemberSpec memberSpec = memberEntry.getValue();
-
- Map<String, Set<Integer>> activeTasks = new
HashMap<>(memberSpec.activeTasks());
+ for (final String memberId : groupSpec.memberIds()) {
+ Map<String, Set<Integer>> activeTasks = new
HashMap<>(groupSpec.memberAssignmentState(memberId).activeTasks());
Review Comment:
Why `new HashMap()` ? -- What's the reason for requiring a copy of the map?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/MemberMetadataAndAssignmentImpl.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.streams.assignor;
+
+import
org.apache.kafka.coordinator.group.api.streams.assignor.MemberAssignmentMetadata;
+import
org.apache.kafka.coordinator.group.api.streams.assignor.MemberAssignmentState;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+/**
+ * Implementation of both the {@link MemberAssignmentMetadata} and the {@link
MemberAssignmentState}
+ * interfaces for a streams group member.
+ *
+ * @param instanceId The instance ID if provided.
+ * @param rackId The rack ID if provided.
+ * @param activeTasks Current active tasks.
+ * @param standbyTasks Current standby tasks.
+ * @param warmupTasks Current warm-up tasks.
+ * @param processId The process ID.
+ * @param clientTags The client tags for a rack-aware assignment.
+ * @param taskOffsets The last received cumulative task offsets of assigned
tasks or dormant tasks.
+ * @param taskEndOffsets The last received cumulative task end offsets of
assigned tasks or dormant tasks.
+ */
+public record MemberMetadataAndAssignmentImpl(Optional<String> instanceId,
+ Optional<String> rackId,
Review Comment:
nit: fix intention (btw, a better formatting would be:
```
public record MemberMetadataAndAssignmentImpl(
Optional<String> instanceId,
[...],
Map<String, Map<Integer, Long>> taskEndOffsets
) implements MemberAssignmentMetadata, MemberAssignmentState {
```
It avoid any indention shifts if we would rename the `record`
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -98,33 +108,33 @@ private void initialize(final GroupSpec groupSpec, final
TopologyDescriber topol
if (topologyDescriber.isStateful(subtopology))
localState.totalTasks += numberOfPartitions *
localState.numStandbyReplicas;
}
- localState.totalMembersWithActiveTaskCapacity =
groupSpec.members().size();
- localState.totalMembersWithTaskCapacity = groupSpec.members().size();
+ localState.totalMembersWithActiveTaskCapacity =
groupSpec.memberIds().size();
+ localState.totalMembersWithTaskCapacity = groupSpec.memberIds().size();
localState.activeTasksPerMember =
computeTasksPerMember(localState.totalActiveTasks,
localState.totalMembersWithActiveTaskCapacity);
localState.totalTasksPerMember =
computeTasksPerMember(localState.totalTasks,
localState.totalMembersWithTaskCapacity);
localState.processIdToState = new
HashMap<>(localState.totalMembersWithActiveTaskCapacity);
localState.activeTaskToPrevMember = new
HashMap<>(localState.totalActiveTasks);
localState.standbyTaskToPrevMember = new
HashMap<>(localState.numStandbyReplicas > 0 ? (localState.totalTasks -
localState.totalActiveTasks) / localState.numStandbyReplicas : 0);
- for (final Map.Entry<String, AssignmentMemberSpec> memberEntry :
groupSpec.members().entrySet()) {
- final String memberId = memberEntry.getKey();
- final String processId = memberEntry.getValue().processId();
+ for (final String memberId : groupSpec.memberIds()) {
+ final MemberAssignmentMetadata memberMetadata =
groupSpec.memberMetadata(memberId);
Review Comment:
Seems we don't really need `memberMetadata` as variable?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java:
##########
@@ -116,17 +117,19 @@ public TargetAssignmentBuilder(
this.assignmentConfigs = Objects.requireNonNull(assignmentConfigs);
}
- static AssignmentMemberSpec createAssignmentMemberSpec(
+ static MemberMetadataAndAssignmentImpl createMemberMetadataAndAssignment(
StreamsGroupMember member,
TasksTuple targetAssignment,
MemberTaskOffsets taskOffsets
) {
- return new AssignmentMemberSpec(
+ return new MemberMetadataAndAssignmentImpl(
member.instanceId(),
member.rackId(),
targetAssignment.activeTasks(),
targetAssignment.standbyTasks(),
Review Comment:
It seems odd to me, that we are passing the `targetAssignment` data here...
This sounds like a bug to me? I believe we should pass in "current assignment"
instead.
\cc @squah-confluent -- looking into code for KIP-848, it seems we are also
passing the target assignment. Wondering why?
```
// Prepare the member spec for all members.
members.forEach((memberId, member) ->
memberSpecs.put(memberId, newMemberSubscriptionAndAssignment(
member,
targetAssignment.getOrDefault(memberId, Assignment.EMPTY),
topicResolver
))
);
```
##########
group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/GroupSpec.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.api.streams.assignor;
+
+import org.apache.kafka.common.annotation.InterfaceAudience;
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.Collection;
+import java.util.Map;
+
+/**
+ * The group metadata specifications required to compute the target assignment.
+ */
[email protected]
[email protected]
+public interface GroupSpec {
+
+ /**
+ * @return The member Ids of all members in the group.
+ */
+ Collection<String> memberIds();
Review Comment:
Wondering if we should change this to `Set`? -- We know that there won't be
any duplicates... (would be a small KIP change)
\cc @squah-confluent
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignorTest.java:
##########
@@ -1442,9 +1446,9 @@ private AssignmentMemberSpec
createAssignmentMemberSpec(final String processId)
Map.of());
}
- private AssignmentMemberSpec createAssignmentMemberSpec(final String
processId, final Map<String, Set<Integer>> prevActiveTasks,
+ private MemberMetadataAndAssignmentImpl createMemberMetadata(final String
processId, final Map<String, Set<Integer>> prevActiveTasks,
final Map<String,
Set<Integer>> prevStandbyTasks) {
Review Comment:
fit indention -- same comment as elsewhere: better formatting would be:
```
private MemberMetadataAndAssignmentImpl createMemberMetadata(
final String processId,
final Map<String, Set<Integer>> prevActiveTasks,
final Map<String, Set<Integer>> prevStandbyTasks
) {
```
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/MockTaskAssignor.java:
##########
@@ -48,8 +49,8 @@ public void prepareGroupAssignment(Map<String, TasksTuple>
memberAssignments) {
Entry::getKey,
entry -> {
TasksTuple tasksTuple = entry.getValue();
- return new MemberAssignment(
- tasksTuple.activeTasks(),
tasksTuple.standbyTasks(), tasksTuple.warmupTasks());
+ return (MemberAssignment) new MemberAssignmentImpl(
Review Comment:
`MemberAssignmentImpl implements MemberAssignment`, so this cast should not
be necessary?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilderTest.java:
##########
@@ -404,6 +438,7 @@ public void addGroupMember(
memberBuilder.setUserEndpoint(new
StreamsGroupMemberMetadataValue.Endpoint().setHost("host").setPort(9090));
memberBuilder.setInstanceId(null);
memberBuilder.setRackId(null);
+ memberBuilder.setAssignedTasks(TasksTupleWithEpochs.EMPTY);
Review Comment:
same
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilderTest.java:
##########
@@ -98,54 +100,80 @@ public void testCreateAssignmentMemberSpec(TaskRole
taskRole) {
.setInstanceId("instanceId")
.setProcessId("processId")
.setClientTags(clientTags)
+ .setAssignedTasks(TasksTupleWithEpochs.EMPTY)
Review Comment:
Why do we need this one now?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java:
##########
@@ -116,17 +117,19 @@ public TargetAssignmentBuilder(
this.assignmentConfigs = Objects.requireNonNull(assignmentConfigs);
}
- static AssignmentMemberSpec createAssignmentMemberSpec(
+ static MemberMetadataAndAssignmentImpl createMemberMetadataAndAssignment(
StreamsGroupMember member,
TasksTuple targetAssignment,
MemberTaskOffsets taskOffsets
) {
- return new AssignmentMemberSpec(
+ return new MemberMetadataAndAssignmentImpl(
member.instanceId(),
member.rackId(),
targetAssignment.activeTasks(),
targetAssignment.standbyTasks(),
Review Comment:
Looking into `StickyTaskAssignor` we use it in `initialize()` to compute
"prev active tasks" and "prev standby tasks" but if we did not reconcile to the
target assignment yet, we would be "sticky" for no reason...
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
