@Test public void testWakeupAfterSyncGroupSentExternalCompletion() throws Exception { mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE)); mockClient.prepareResponse(joinGroupFollowerResponse(1, "memberId", "leaderId", Errors.NONE)); mockClient.prepareResponse(new MockClient.RequestMatcher() { private int invocations = 0; @Override public boolean matches(AbstractRequest body) { invocations++; boolean isSyncGroupRequest = body instanceof SyncGroupRequest; if (isSyncGroupRequest && invocations == 1) throw new WakeupException(); return isSyncGroupRequest; } }, syncGroupResponse(Errors.NONE)); AtomicBoolean heartbeatReceived = prepareFirstHeartbeat(); try { coordinator.ensureActiveGroup(); fail("Should have woken up from ensureActiveGroup()"); } catch (WakeupException e) { } assertEquals(1, coordinator.onJoinPrepareInvokes); assertEquals(0, coordinator.onJoinCompleteInvokes); assertFalse(heartbeatReceived.get()); consumerClient.poll(0); coordinator.ensureActiveGroup(); assertEquals(1, coordinator.onJoinPrepareInvokes); assertEquals(1, coordinator.onJoinCompleteInvokes); awaitFirstHeartbeat(heartbeatReceived); }
public void ensureActiveGroup() { ensureCoordinatorReady(); startHeartbeatThreadIfNeeded(); joinGroupIfNeeded(); }
AbstractCoordinator implements Closeable { public void ensureActiveGroup() { ensureCoordinatorReady(); startHeartbeatThreadIfNeeded(); joinGroupIfNeeded(); } }
AbstractCoordinator implements Closeable { public void ensureActiveGroup() { ensureCoordinatorReady(); startHeartbeatThreadIfNeeded(); joinGroupIfNeeded(); } AbstractCoordinator(ConsumerNetworkClient client,
                               String groupId,
                               int rebalanceTimeoutMs,
                               int sessionTimeoutMs,
                               int heartbeatIntervalMs,
                               Metrics metrics,
                               String metricGrpPrefix,
                               Time time,
                               long retryBackoffMs,
                               boolean leaveGroupOnClose); }
AbstractCoordinator implements Closeable { public void ensureActiveGroup() { ensureCoordinatorReady(); startHeartbeatThreadIfNeeded(); joinGroupIfNeeded(); } AbstractCoordinator(ConsumerNetworkClient client,
                               String groupId,
                               int rebalanceTimeoutMs,
                               int sessionTimeoutMs,
                               int heartbeatIntervalMs,
                               Metrics metrics,
                               String metricGrpPrefix,
                               Time time,
                               long retryBackoffMs,
                               boolean leaveGroupOnClose); synchronized void ensureCoordinatorReady(); void ensureActiveGroup(); boolean coordinatorUnknown(); @Override final void close(); synchronized void maybeLeaveGroup(); }
AbstractCoordinator implements Closeable { public void ensureActiveGroup() { ensureCoordinatorReady(); startHeartbeatThreadIfNeeded(); joinGroupIfNeeded(); } AbstractCoordinator(ConsumerNetworkClient client,
                               String groupId,
                               int rebalanceTimeoutMs,
                               int sessionTimeoutMs,
                               int heartbeatIntervalMs,
                               Metrics metrics,
                               String metricGrpPrefix,
                               Time time,
                               long retryBackoffMs,
                               boolean leaveGroupOnClose); synchronized void ensureCoordinatorReady(); void ensureActiveGroup(); boolean coordinatorUnknown(); @Override final void close(); synchronized void maybeLeaveGroup();  }