@Test public void testJoinLeaderCannotAssign() { EasyMock.expect(configStorage.snapshot()).andReturn(configState1); EasyMock.expect(configStorage.snapshot()).andReturn(configState2); PowerMock.replayAll(); final String memberId = "member"; client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE)); coordinator.ensureCoordinatorReady(); client.prepareResponse(joinGroupFollowerResponse(1, memberId, "leader", Errors.NONE)); MockClient.RequestMatcher matcher = new MockClient.RequestMatcher() { @Override public boolean matches(AbstractRequest body) { SyncGroupRequest sync = (SyncGroupRequest) body; return sync.memberId().equals(memberId) && sync.generationId() == 1 && sync.groupAssignment().isEmpty(); } }; client.prepareResponse(matcher, syncGroupResponse(ConnectProtocol.Assignment.CONFIG_MISMATCH, "leader", 10L, Collections.<String>emptyList(), Collections.<ConnectorTaskId>emptyList(), Errors.NONE)); client.prepareResponse(joinGroupFollowerResponse(1, memberId, "leader", Errors.NONE)); client.prepareResponse(matcher, syncGroupResponse(ConnectProtocol.Assignment.NO_ERROR, "leader", 1L, Collections.<String>emptyList(), Collections.singletonList(taskId1x0), Errors.NONE)); coordinator.ensureActiveGroup(); PowerMock.verifyAll(); }
public String memberId() { Generation generation = generation(); if (generation != null) return generation.memberId; return JoinGroupRequest.UNKNOWN_MEMBER_ID; }
WorkerCoordinator extends AbstractCoordinator implements Closeable { public String memberId() { Generation generation = generation(); if (generation != null) return generation.memberId; return JoinGroupRequest.UNKNOWN_MEMBER_ID; } }
WorkerCoordinator extends AbstractCoordinator implements Closeable { public String memberId() { Generation generation = generation(); if (generation != null) return generation.memberId; return JoinGroupRequest.UNKNOWN_MEMBER_ID; } 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 String memberId() { Generation generation = generation(); if (generation != null) return generation.memberId; return JoinGroupRequest.UNKNOWN_MEMBER_ID; } 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 String memberId() { Generation generation = generation(); if (generation != null) return generation.memberId; return JoinGroupRequest.UNKNOWN_MEMBER_ID; } 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; }