kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From j...@apache.org
Subject kafka git commit: KAFKA-4553; Improve round robin assignment in Connect to avoid uneven distributions of connectors and tasks
Date Mon, 19 Dec 2016 20:40:03 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 65acff32d -> bc9ef716a


KAFKA-4553; Improve round robin assignment in Connect to avoid uneven distributions of connectors
and tasks

Author: Ewen Cheslack-Postava <me@ewencp.org>

Reviewers: Konstantine Karantasis <konstantine@confluent.io>, Jason Gustafson <jason@confluent.io>

Closes #2272 from ewencp/kafka-4553-better-connect-round-robin


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/bc9ef716
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/bc9ef716
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/bc9ef716

Branch: refs/heads/trunk
Commit: bc9ef716af8ac387d63b22ee973a9fc6f864bebb
Parents: 65acff3
Author: Ewen Cheslack-Postava <me@ewencp.org>
Authored: Mon Dec 19 12:37:58 2016 -0800
Committer: Jason Gustafson <jason@confluent.io>
Committed: Mon Dec 19 12:37:58 2016 -0800

----------------------------------------------------------------------
 .../runtime/distributed/WorkerCoordinator.java  |  38 +++---
 .../distributed/WorkerCoordinatorTest.java      | 134 ++++++++++++++-----
 2 files changed, 123 insertions(+), 49 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/bc9ef716/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java
----------------------------------------------------------------------
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java
b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java
index 88a0a8d..58525c5 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java
@@ -154,25 +154,25 @@ public final class WorkerCoordinator extends AbstractCoordinator implements
Clos
     protected Map<String, ByteBuffer> performAssignment(String leaderId, String protocol,
Map<String, ByteBuffer> allMemberMetadata) {
         log.debug("Performing task assignment");
 
-        Map<String, ConnectProtocol.WorkerState> allConfigs = new HashMap<>();
+        Map<String, ConnectProtocol.WorkerState> memberConfigs = new HashMap<>();
         for (Map.Entry<String, ByteBuffer> entry : allMemberMetadata.entrySet())
-            allConfigs.put(entry.getKey(), ConnectProtocol.deserializeMetadata(entry.getValue()));
+            memberConfigs.put(entry.getKey(), ConnectProtocol.deserializeMetadata(entry.getValue()));
 
-        long maxOffset = findMaxMemberConfigOffset(allConfigs);
+        long maxOffset = findMaxMemberConfigOffset(memberConfigs);
         Long leaderOffset = ensureLeaderConfig(maxOffset);
         if (leaderOffset == null)
-            return fillAssignmentsAndSerialize(allConfigs.keySet(), ConnectProtocol.Assignment.CONFIG_MISMATCH,
-                    leaderId, allConfigs.get(leaderId).url(), maxOffset,
+            return fillAssignmentsAndSerialize(memberConfigs.keySet(), ConnectProtocol.Assignment.CONFIG_MISMATCH,
+                    leaderId, memberConfigs.get(leaderId).url(), maxOffset,
                     new HashMap<String, List<String>>(), new HashMap<String,
List<ConnectorTaskId>>());
-        return performTaskAssignment(leaderId, leaderOffset, allConfigs);
+        return performTaskAssignment(leaderId, leaderOffset, memberConfigs);
     }
 
-    private long findMaxMemberConfigOffset(Map<String, ConnectProtocol.WorkerState>
allConfigs) {
+    private long findMaxMemberConfigOffset(Map<String, ConnectProtocol.WorkerState>
memberConfigs) {
         // The new config offset is the maximum seen by any member. We always perform assignment
using this offset,
         // even if some members have fallen behind. The config offset used to generate the
assignment is included in
         // the response so members that have fallen behind will not use the assignment until
they have caught up.
         Long maxOffset = null;
-        for (Map.Entry<String, ConnectProtocol.WorkerState> stateEntry : allConfigs.entrySet())
{
+        for (Map.Entry<String, ConnectProtocol.WorkerState> stateEntry : memberConfigs.entrySet())
{
             long memberRootOffset = stateEntry.getValue().offset();
             if (maxOffset == null)
                 maxOffset = memberRootOffset;
@@ -205,13 +205,18 @@ public final class WorkerCoordinator extends AbstractCoordinator implements
Clos
         return maxOffset;
     }
 
-    private Map<String, ByteBuffer> performTaskAssignment(String leaderId, long maxOffset,
Map<String, ConnectProtocol.WorkerState> allConfigs) {
+    private Map<String, ByteBuffer> performTaskAssignment(String leaderId, long maxOffset,
Map<String, ConnectProtocol.WorkerState> memberConfigs) {
         Map<String, List<String>> connectorAssignments = new HashMap<>();
         Map<String, List<ConnectorTaskId>> taskAssignments = new HashMap<>();
 
-        // Perform round-robin task assignment
-        CircularIterator<String> memberIt = new CircularIterator<>(sorted(allConfigs.keySet()));
-        for (String connectorId : sorted(configSnapshot.connectors())) {
+        // Perform round-robin task assignment. Assign all connectors and then all tasks
because assigning both the
+        // connector and its tasks can lead to very uneven distribution of work in some common
cases (e.g. for connectors
+        // that generate only 1 task each; in a cluster of 2 or an even # of nodes, only
even nodes will be assigned
+        // connectors and only odd nodes will be assigned tasks, but tasks are, on average,
actually more resource
+        // intensive than connectors).
+        List<String> connectorsSorted = sorted(configSnapshot.connectors());
+        CircularIterator<String> memberIt = new CircularIterator<>(sorted(memberConfigs.keySet()));
+        for (String connectorId : connectorsSorted) {
             String connectorAssignedTo = memberIt.next();
             log.trace("Assigning connector {} to {}", connectorId, connectorAssignedTo);
             List<String> memberConnectors = connectorAssignments.get(connectorAssignedTo);
@@ -220,7 +225,8 @@ public final class WorkerCoordinator extends AbstractCoordinator implements
Clos
                 connectorAssignments.put(connectorAssignedTo, memberConnectors);
             }
             memberConnectors.add(connectorId);
-
+        }
+        for (String connectorId : connectorsSorted) {
             for (ConnectorTaskId taskId : sorted(configSnapshot.tasks(connectorId))) {
                 String taskAssignedTo = memberIt.next();
                 log.trace("Assigning task {} to {}", taskId, taskAssignedTo);
@@ -233,10 +239,10 @@ public final class WorkerCoordinator extends AbstractCoordinator implements
Clos
             }
         }
 
-        this.leaderState = new LeaderState(allConfigs, connectorAssignments, taskAssignments);
+        this.leaderState = new LeaderState(memberConfigs, connectorAssignments, taskAssignments);
 
-        return fillAssignmentsAndSerialize(allConfigs.keySet(), ConnectProtocol.Assignment.NO_ERROR,
-                leaderId, allConfigs.get(leaderId).url(), maxOffset, connectorAssignments,
taskAssignments);
+        return fillAssignmentsAndSerialize(memberConfigs.keySet(), ConnectProtocol.Assignment.NO_ERROR,
+                leaderId, memberConfigs.get(leaderId).url(), maxOffset, connectorAssignments,
taskAssignments);
     }
 
     private Map<String, ByteBuffer> fillAssignmentsAndSerialize(Collection<String>
members,

http://git-wip-us.apache.org/repos/asf/kafka/blob/bc9ef716/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinatorTest.java
----------------------------------------------------------------------
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinatorTest.java
b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinatorTest.java
index 3aff8f2..92393a1 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinatorTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinatorTest.java
@@ -58,11 +58,13 @@ public class WorkerCoordinatorTest {
     private static final String LEADER_URL = "leaderUrl:8083";
     private static final String MEMBER_URL = "memberUrl:8083";
 
-    private String connectorId = "connector";
+    private String connectorId1 = "connector1";
     private String connectorId2 = "connector2";
-    private ConnectorTaskId taskId0 = new ConnectorTaskId(connectorId, 0);
-    private ConnectorTaskId taskId1 = new ConnectorTaskId(connectorId, 1);
-    private ConnectorTaskId taskId2 = new ConnectorTaskId(connectorId2, 0);
+    private String connectorId3 = "connector3";
+    private ConnectorTaskId taskId1x0 = new ConnectorTaskId(connectorId1, 0);
+    private ConnectorTaskId taskId1x1 = new ConnectorTaskId(connectorId1, 1);
+    private ConnectorTaskId taskId2x0 = new ConnectorTaskId(connectorId2, 0);
+    private ConnectorTaskId taskId3x0 = new ConnectorTaskId(connectorId3, 0);
 
     private String groupId = "test-group";
     private int sessionTimeoutMs = 10;
@@ -82,6 +84,7 @@ public class WorkerCoordinatorTest {
 
     private ClusterConfigState configState1;
     private ClusterConfigState configState2;
+    private ClusterConfigState configStateSingleTaskConnectors;
 
     @Before
     public void setup() {
@@ -110,32 +113,60 @@ public class WorkerCoordinatorTest {
                 rebalanceListener);
 
         configState1 = new ClusterConfigState(
-                1L, Collections.singletonMap(connectorId, 1),
-                Collections.singletonMap(connectorId, (Map<String, String>) new HashMap<String,
String>()),
-                Collections.singletonMap(connectorId, TargetState.STARTED),
-                Collections.singletonMap(taskId0, (Map<String, String>) new HashMap<String,
String>()),
+                1L,
+                Collections.singletonMap(connectorId1, 1),
+                Collections.singletonMap(connectorId1, (Map<String, String>) new HashMap<String,
String>()),
+                Collections.singletonMap(connectorId1, TargetState.STARTED),
+                Collections.singletonMap(taskId1x0, (Map<String, String>) new HashMap<String,
String>()),
                 Collections.<String>emptySet()
         );
+
         Map<String, Integer> configState2ConnectorTaskCounts = new HashMap<>();
-        configState2ConnectorTaskCounts.put(connectorId, 2);
+        configState2ConnectorTaskCounts.put(connectorId1, 2);
         configState2ConnectorTaskCounts.put(connectorId2, 1);
         Map<String, Map<String, String>> configState2ConnectorConfigs = new HashMap<>();
-        configState2ConnectorConfigs.put(connectorId, new HashMap<String, String>());
+        configState2ConnectorConfigs.put(connectorId1, new HashMap<String, String>());
         configState2ConnectorConfigs.put(connectorId2, new HashMap<String, String>());
-        Map<String, TargetState> targetStates = new HashMap<>();
-        targetStates.put(connectorId, TargetState.STARTED);
-        targetStates.put(connectorId2, TargetState.STARTED);
+        Map<String, TargetState> configState2TargetStates = new HashMap<>();
+        configState2TargetStates.put(connectorId1, TargetState.STARTED);
+        configState2TargetStates.put(connectorId2, TargetState.STARTED);
         Map<ConnectorTaskId, Map<String, String>> configState2TaskConfigs = new
HashMap<>();
-        configState2TaskConfigs.put(taskId0, new HashMap<String, String>());
-        configState2TaskConfigs.put(taskId1, new HashMap<String, String>());
-        configState2TaskConfigs.put(taskId2, new HashMap<String, String>());
+        configState2TaskConfigs.put(taskId1x0, new HashMap<String, String>());
+        configState2TaskConfigs.put(taskId1x1, new HashMap<String, String>());
+        configState2TaskConfigs.put(taskId2x0, new HashMap<String, String>());
         configState2 = new ClusterConfigState(
-                2L, configState2ConnectorTaskCounts,
+                2L,
+                configState2ConnectorTaskCounts,
                 configState2ConnectorConfigs,
-                targetStates,
+                configState2TargetStates,
                 configState2TaskConfigs,
                 Collections.<String>emptySet()
         );
+
+        Map<String, Integer> configStateSingleTaskConnectorsConnectorTaskCounts = new
HashMap<>();
+        configStateSingleTaskConnectorsConnectorTaskCounts.put(connectorId1, 1);
+        configStateSingleTaskConnectorsConnectorTaskCounts.put(connectorId2, 1);
+        configStateSingleTaskConnectorsConnectorTaskCounts.put(connectorId3, 1);
+        Map<String, Map<String, String>> configStateSingleTaskConnectorsConnectorConfigs
= new HashMap<>();
+        configStateSingleTaskConnectorsConnectorConfigs.put(connectorId1, new HashMap<String,
String>());
+        configStateSingleTaskConnectorsConnectorConfigs.put(connectorId2, new HashMap<String,
String>());
+        configStateSingleTaskConnectorsConnectorConfigs.put(connectorId3, new HashMap<String,
String>());
+        Map<String, TargetState> configStateSingleTaskConnectorsTargetStates = new
HashMap<>();
+        configStateSingleTaskConnectorsTargetStates.put(connectorId1, TargetState.STARTED);
+        configStateSingleTaskConnectorsTargetStates.put(connectorId2, TargetState.STARTED);
+        configStateSingleTaskConnectorsTargetStates.put(connectorId3, TargetState.STARTED);
+        Map<ConnectorTaskId, Map<String, String>> configStateSingleTaskConnectorsTaskConfigs
= new HashMap<>();
+        configStateSingleTaskConnectorsTaskConfigs.put(taskId1x0, new HashMap<String,
String>());
+        configStateSingleTaskConnectorsTaskConfigs.put(taskId2x0, new HashMap<String,
String>());
+        configStateSingleTaskConnectorsTaskConfigs.put(taskId3x0, new HashMap<String,
String>());
+        configStateSingleTaskConnectors = new ClusterConfigState(
+                2L,
+                configStateSingleTaskConnectorsConnectorTaskCounts,
+                configStateSingleTaskConnectorsConnectorConfigs,
+                configStateSingleTaskConnectorsTargetStates,
+                configStateSingleTaskConnectorsTaskConfigs,
+                Collections.<String>emptySet()
+        );
     }
 
     @After
@@ -187,7 +218,7 @@ public class WorkerCoordinatorTest {
                         sync.generationId() == 1 &&
                         sync.groupAssignment().containsKey(consumerId);
             }
-        }, syncGroupResponse(ConnectProtocol.Assignment.NO_ERROR, "leader", 1L, Collections.singletonList(connectorId),
+        }, syncGroupResponse(ConnectProtocol.Assignment.NO_ERROR, "leader", 1L, Collections.singletonList(connectorId1),
                 Collections.<ConnectorTaskId>emptyList(), Errors.NONE.code()));
         coordinator.ensureActiveGroup();
 
@@ -197,7 +228,7 @@ public class WorkerCoordinatorTest {
         assertFalse(rebalanceListener.assignment.failed());
         assertEquals(1L, rebalanceListener.assignment.offset());
         assertEquals("leader", rebalanceListener.assignment.leader());
-        assertEquals(Collections.singletonList(connectorId), rebalanceListener.assignment.connectors());
+        assertEquals(Collections.singletonList(connectorId1), rebalanceListener.assignment.connectors());
         assertEquals(Collections.emptyList(), rebalanceListener.assignment.tasks());
 
         PowerMock.verifyAll();
@@ -225,7 +256,7 @@ public class WorkerCoordinatorTest {
                         sync.groupAssignment().isEmpty();
             }
         }, syncGroupResponse(ConnectProtocol.Assignment.NO_ERROR, "leader", 1L, Collections.<String>emptyList(),
-                Collections.singletonList(taskId0), Errors.NONE.code()));
+                Collections.singletonList(taskId1x0), Errors.NONE.code()));
         coordinator.ensureActiveGroup();
 
         assertFalse(coordinator.needRejoin());
@@ -234,7 +265,7 @@ public class WorkerCoordinatorTest {
         assertFalse(rebalanceListener.assignment.failed());
         assertEquals(1L, rebalanceListener.assignment.offset());
         assertEquals(Collections.emptyList(), rebalanceListener.assignment.connectors());
-        assertEquals(Collections.singletonList(taskId0), rebalanceListener.assignment.tasks());
+        assertEquals(Collections.singletonList(taskId1x0), rebalanceListener.assignment.tasks());
 
         PowerMock.verifyAll();
     }
@@ -270,7 +301,7 @@ public class WorkerCoordinatorTest {
                 Collections.<String>emptyList(), Collections.<ConnectorTaskId>emptyList(),
Errors.NONE.code()));
         client.prepareResponse(joinGroupFollowerResponse(1, memberId, "leader", Errors.NONE.code()));
         client.prepareResponse(matcher, syncGroupResponse(ConnectProtocol.Assignment.NO_ERROR,
"leader", 1L,
-                Collections.<String>emptyList(), Collections.singletonList(taskId0),
Errors.NONE.code()));
+                Collections.<String>emptyList(), Collections.singletonList(taskId1x0),
Errors.NONE.code()));
         coordinator.ensureActiveGroup();
 
         PowerMock.verifyAll();
@@ -289,7 +320,7 @@ public class WorkerCoordinatorTest {
         // join the group once
         client.prepareResponse(joinGroupFollowerResponse(1, "consumer", "leader", Errors.NONE.code()));
         client.prepareResponse(syncGroupResponse(ConnectProtocol.Assignment.NO_ERROR, "leader",
1L, Collections.<String>emptyList(),
-                Collections.singletonList(taskId0), Errors.NONE.code()));
+                Collections.singletonList(taskId1x0), Errors.NONE.code()));
         coordinator.ensureActiveGroup();
 
         assertEquals(0, rebalanceListener.revokedCount);
@@ -297,22 +328,22 @@ public class WorkerCoordinatorTest {
         assertFalse(rebalanceListener.assignment.failed());
         assertEquals(1L, rebalanceListener.assignment.offset());
         assertEquals(Collections.emptyList(), rebalanceListener.assignment.connectors());
-        assertEquals(Collections.singletonList(taskId0), rebalanceListener.assignment.tasks());
+        assertEquals(Collections.singletonList(taskId1x0), rebalanceListener.assignment.tasks());
 
         // and join the group again
         coordinator.requestRejoin();
         client.prepareResponse(joinGroupFollowerResponse(1, "consumer", "leader", Errors.NONE.code()));
-        client.prepareResponse(syncGroupResponse(ConnectProtocol.Assignment.NO_ERROR, "leader",
1L, Collections.singletonList(connectorId),
+        client.prepareResponse(syncGroupResponse(ConnectProtocol.Assignment.NO_ERROR, "leader",
1L, Collections.singletonList(connectorId1),
                 Collections.<ConnectorTaskId>emptyList(), Errors.NONE.code()));
         coordinator.ensureActiveGroup();
 
         assertEquals(1, rebalanceListener.revokedCount);
         assertEquals(Collections.emptyList(), rebalanceListener.revokedConnectors);
-        assertEquals(Collections.singletonList(taskId0), rebalanceListener.revokedTasks);
+        assertEquals(Collections.singletonList(taskId1x0), rebalanceListener.revokedTasks);
         assertEquals(2, rebalanceListener.assignedCount);
         assertFalse(rebalanceListener.assignment.failed());
         assertEquals(1L, rebalanceListener.assignment.offset());
-        assertEquals(Collections.singletonList(connectorId), rebalanceListener.assignment.connectors());
+        assertEquals(Collections.singletonList(connectorId1), rebalanceListener.assignment.connectors());
         assertEquals(Collections.emptyList(), rebalanceListener.assignment.tasks());
 
         PowerMock.verifyAll();
@@ -341,7 +372,7 @@ public class WorkerCoordinatorTest {
         assertEquals(false, leaderAssignment.failed());
         assertEquals("leader", leaderAssignment.leader());
         assertEquals(1, leaderAssignment.offset());
-        assertEquals(Collections.singletonList(connectorId), leaderAssignment.connectors());
+        assertEquals(Collections.singletonList(connectorId1), leaderAssignment.connectors());
         assertEquals(Collections.emptyList(), leaderAssignment.tasks());
 
         ConnectProtocol.Assignment memberAssignment = ConnectProtocol.deserializeAssignment(result.get("member"));
@@ -349,7 +380,7 @@ public class WorkerCoordinatorTest {
         assertEquals("leader", memberAssignment.leader());
         assertEquals(1, memberAssignment.offset());
         assertEquals(Collections.emptyList(), memberAssignment.connectors());
-        assertEquals(Collections.singletonList(taskId0), memberAssignment.tasks());
+        assertEquals(Collections.singletonList(taskId1x0), memberAssignment.tasks());
 
         PowerMock.verifyAll();
     }
@@ -377,15 +408,52 @@ public class WorkerCoordinatorTest {
         assertEquals(false, leaderAssignment.failed());
         assertEquals("leader", leaderAssignment.leader());
         assertEquals(1, leaderAssignment.offset());
-        assertEquals(Collections.singletonList(connectorId), leaderAssignment.connectors());
-        assertEquals(Arrays.asList(taskId1, taskId2), leaderAssignment.tasks());
+        assertEquals(Collections.singletonList(connectorId1), leaderAssignment.connectors());
+        assertEquals(Arrays.asList(taskId1x0, taskId2x0), leaderAssignment.tasks());
+
+        ConnectProtocol.Assignment memberAssignment = ConnectProtocol.deserializeAssignment(result.get("member"));
+        assertEquals(false, memberAssignment.failed());
+        assertEquals("leader", memberAssignment.leader());
+        assertEquals(1, memberAssignment.offset());
+        assertEquals(Collections.singletonList(connectorId2), memberAssignment.connectors());
+        assertEquals(Collections.singletonList(taskId1x1), memberAssignment.tasks());
+
+        PowerMock.verifyAll();
+    }
+
+    @Test
+    public void testLeaderPerformAssignmentSingleTaskConnectors() throws Exception {
+        // Since all the protocol responses are mocked, the other tests validate doSync runs,
but don't validate its
+        // output. So we test it directly here.
+
+        EasyMock.expect(configStorage.snapshot()).andReturn(configStateSingleTaskConnectors);
+
+        PowerMock.replayAll();
+
+        // Prime the current configuration state
+        coordinator.metadata();
+
+        Map<String, ByteBuffer> configs = new HashMap<>();
+        // Mark everyone as in sync with configState1
+        configs.put("leader", ConnectProtocol.serializeMetadata(new ConnectProtocol.WorkerState(LEADER_URL,
1L)));
+        configs.put("member", ConnectProtocol.serializeMetadata(new ConnectProtocol.WorkerState(MEMBER_URL,
1L)));
+        Map<String, ByteBuffer> result = Whitebox.invokeMethod(coordinator, "performAssignment",
"leader", WorkerCoordinator.DEFAULT_SUBPROTOCOL, configs);
+
+        // Round robin assignment when there are the same number of connectors and tasks
should result in each being
+        // evenly distributed across the workers, i.e. round robin assignment of connectors
first, then followed by tasks
+        ConnectProtocol.Assignment leaderAssignment = ConnectProtocol.deserializeAssignment(result.get("leader"));
+        assertEquals(false, leaderAssignment.failed());
+        assertEquals("leader", leaderAssignment.leader());
+        assertEquals(1, leaderAssignment.offset());
+        assertEquals(Arrays.asList(connectorId1, connectorId3), leaderAssignment.connectors());
+        assertEquals(Arrays.asList(taskId2x0), leaderAssignment.tasks());
 
         ConnectProtocol.Assignment memberAssignment = ConnectProtocol.deserializeAssignment(result.get("member"));
         assertEquals(false, memberAssignment.failed());
         assertEquals("leader", memberAssignment.leader());
         assertEquals(1, memberAssignment.offset());
         assertEquals(Collections.singletonList(connectorId2), memberAssignment.connectors());
-        assertEquals(Collections.singletonList(taskId0), memberAssignment.tasks());
+        assertEquals(Arrays.asList(taskId1x0, taskId3x0), memberAssignment.tasks());
 
         PowerMock.verifyAll();
     }


Mime
View raw message