WorkerCoordinator extends AbstractCoordinator implements Closeable { @Override public List<ProtocolMetadata> metadata() { configSnapshot = configStorage.snapshot(); ConnectProtocol.WorkerState workerState = new ConnectProtocol.WorkerState(restUrl, configSnapshot.offset()); ByteBuffer metadata = ConnectProtocol.serializeMetadata(workerState); return Collections.singletonList(new ProtocolMetadata(DEFAULT_SUBPROTOCOL, metadata)); } WorkerCoordinator(ConsumerNetworkClient client,
                             String groupId,
                             int sessionTimeoutMs,
                             int heartbeatIntervalMs,
                             Metrics metrics,
                             String metricGrpPrefix,
                             Time time,
                             long retryBackoffMs,
                             String restUrl,
                             ConfigBackingStore configStorage,
                             WorkerRebalanceListener listener); void requestRejoin(); @Override String protocolType(); @Override List<ProtocolMetadata> metadata(); @Override boolean needRejoin(); String memberId(); @Override void close(); String ownerUrl(String connector); String ownerUrl(ConnectorTaskId task); static final String DEFAULT_SUBPROTOCOL; }
@Test public void testMetadata() { EasyMock.expect(configStorage.snapshot()).andReturn(configState1); PowerMock.replayAll(); List<ProtocolMetadata> serialized = coordinator.metadata(); assertEquals(1, serialized.size()); ProtocolMetadata defaultMetadata = serialized.get(0); assertEquals(WorkerCoordinator.DEFAULT_SUBPROTOCOL, defaultMetadata.name()); ConnectProtocol.WorkerState state = ConnectProtocol.deserializeMetadata(defaultMetadata.metadata()); assertEquals(1, state.offset()); PowerMock.verifyAll(); }
@Test public void testLeaderPerformAssignment1() throws Exception { EasyMock.expect(configStorage.snapshot()).andReturn(configState1); PowerMock.replayAll(); coordinator.metadata(); Map<String, ByteBuffer> configs = new HashMap<>(); 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); ConnectProtocol.Assignment leaderAssignment = ConnectProtocol.deserializeAssignment(result.get("leader")); assertEquals(false, leaderAssignment.failed()); assertEquals("leader", leaderAssignment.leader()); assertEquals(1, leaderAssignment.offset()); assertEquals(Collections.singletonList(connectorId), leaderAssignment.connectors()); assertEquals(Collections.emptyList(), leaderAssignment.tasks()); ConnectProtocol.Assignment memberAssignment = ConnectProtocol.deserializeAssignment(result.get("member")); assertEquals(false, memberAssignment.failed()); assertEquals("leader", memberAssignment.leader()); assertEquals(1, memberAssignment.offset()); assertEquals(Collections.emptyList(), memberAssignment.connectors()); assertEquals(Collections.singletonList(taskId0), memberAssignment.tasks()); PowerMock.verifyAll(); }
@Test public void testLeaderPerformAssignment2() throws Exception { EasyMock.expect(configStorage.snapshot()).andReturn(configState2); PowerMock.replayAll(); coordinator.metadata(); Map<String, ByteBuffer> configs = new HashMap<>(); 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); ConnectProtocol.Assignment leaderAssignment = ConnectProtocol.deserializeAssignment(result.get("leader")); 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()); 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()); PowerMock.verifyAll(); }