@Test public void testRejoinGroup() { EasyMock.expect(configStorage.snapshot()).andReturn(configState1); EasyMock.expect(configStorage.snapshot()).andReturn(configState1); PowerMock.replayAll(); client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE)); coordinator.ensureCoordinatorReady(); client.prepareResponse(joinGroupFollowerResponse(1, "consumer", "leader", Errors.NONE)); client.prepareResponse(syncGroupResponse(ConnectProtocol.Assignment.NO_ERROR, "leader", 1L, Collections.<String>emptyList(), Collections.singletonList(taskId1x0), Errors.NONE)); coordinator.ensureActiveGroup(); assertEquals(0, rebalanceListener.revokedCount); assertEquals(1, rebalanceListener.assignedCount); assertFalse(rebalanceListener.assignment.failed()); assertEquals(1L, rebalanceListener.assignment.offset()); assertEquals(Collections.emptyList(), rebalanceListener.assignment.connectors()); assertEquals(Collections.singletonList(taskId1x0), rebalanceListener.assignment.tasks()); coordinator.requestRejoin(); client.prepareResponse(joinGroupFollowerResponse(1, "consumer", "leader", Errors.NONE)); client.prepareResponse(syncGroupResponse(ConnectProtocol.Assignment.NO_ERROR, "leader", 1L, Collections.singletonList(connectorId1), Collections.<ConnectorTaskId>emptyList(), Errors.NONE)); coordinator.ensureActiveGroup(); assertEquals(1, rebalanceListener.revokedCount); assertEquals(Collections.emptyList(), rebalanceListener.revokedConnectors); assertEquals(Collections.singletonList(taskId1x0), rebalanceListener.revokedTasks); assertEquals(2, rebalanceListener.assignedCount); assertFalse(rebalanceListener.assignment.failed()); assertEquals(1L, rebalanceListener.assignment.offset()); assertEquals(Collections.singletonList(connectorId1), rebalanceListener.assignment.connectors()); assertEquals(Collections.emptyList(), rebalanceListener.assignment.tasks()); PowerMock.verifyAll(); }
public void requestRejoin() { rejoinRequested = true; }
WorkerCoordinator extends AbstractCoordinator implements Closeable { public void requestRejoin() { rejoinRequested = true; } }
WorkerCoordinator extends AbstractCoordinator implements Closeable { public void requestRejoin() { rejoinRequested = true; } WorkerCoordinator(ConsumerNetworkClient client,
                             String groupId,
                             int rebalanceTimeoutMs,
                             int sessionTimeoutMs,
                             int heartbeatIntervalMs,
                             Metrics metrics,
                             String metricGrpPrefix,
                             Time time,
                             long retryBackoffMs,
                             String restUrl,
                             ConfigBackingStore configStorage,
                             WorkerRebalanceListener listener); }
WorkerCoordinator extends AbstractCoordinator implements Closeable { public void requestRejoin() { rejoinRequested = true; } WorkerCoordinator(ConsumerNetworkClient client,
                             String groupId,
                             int rebalanceTimeoutMs,
                             int sessionTimeoutMs,
                             int heartbeatIntervalMs,
                             Metrics metrics,
                             String metricGrpPrefix,
                             Time time,
                             long retryBackoffMs,
                             String restUrl,
                             ConfigBackingStore configStorage,
                             WorkerRebalanceListener listener); void requestRejoin(); @Override String protocolType(); void poll(long timeout); @Override List<ProtocolMetadata> metadata(); String memberId(); String ownerUrl(String connector); String ownerUrl(ConnectorTaskId task); }
WorkerCoordinator extends AbstractCoordinator implements Closeable { public void requestRejoin() { rejoinRequested = true; } WorkerCoordinator(ConsumerNetworkClient client,
                             String groupId,
                             int rebalanceTimeoutMs,
                             int sessionTimeoutMs,
                             int heartbeatIntervalMs,
                             Metrics metrics,
                             String metricGrpPrefix,
                             Time time,
                             long retryBackoffMs,
                             String restUrl,
                             ConfigBackingStore configStorage,
                             WorkerRebalanceListener listener); void requestRejoin(); @Override String protocolType(); void poll(long timeout); @Override List<ProtocolMetadata> metadata(); String memberId(); String ownerUrl(String connector); String ownerUrl(ConnectorTaskId task); static final String DEFAULT_SUBPROTOCOL; }