@Test public void testLookupCoordinator() throws Exception { mockClient.setNode(null); RequestFuture<Void> noBrokersAvailableFuture = coordinator.lookupCoordinator(); assertTrue("Failed future expected", noBrokersAvailableFuture.failed()); mockClient.setNode(node); RequestFuture<Void> future = coordinator.lookupCoordinator(); assertFalse("Request not sent", future.isDone()); assertTrue("New request sent while one is in progress", future == coordinator.lookupCoordinator()); mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE)); coordinator.ensureCoordinatorReady(); assertTrue("New request not sent after previous completed", future != coordinator.lookupCoordinator()); }
protected synchronized RequestFuture<Void> lookupCoordinator() { if (findCoordinatorFuture == null) { Node node = this.client.leastLoadedNode(); if (node == null) { log.debug("No broker available to send GroupCoordinator request for group {}", groupId); return RequestFuture.noBrokersAvailable(); } else findCoordinatorFuture = sendGroupCoordinatorRequest(node); } return findCoordinatorFuture; }
AbstractCoordinator implements Closeable { protected synchronized RequestFuture<Void> lookupCoordinator() { if (findCoordinatorFuture == null) { Node node = this.client.leastLoadedNode(); if (node == null) { log.debug("No broker available to send GroupCoordinator request for group {}", groupId); return RequestFuture.noBrokersAvailable(); } else findCoordinatorFuture = sendGroupCoordinatorRequest(node); } return findCoordinatorFuture; } }
AbstractCoordinator implements Closeable { protected synchronized RequestFuture<Void> lookupCoordinator() { if (findCoordinatorFuture == null) { Node node = this.client.leastLoadedNode(); if (node == null) { log.debug("No broker available to send GroupCoordinator request for group {}", groupId); return RequestFuture.noBrokersAvailable(); } else findCoordinatorFuture = sendGroupCoordinatorRequest(node); } return findCoordinatorFuture; } AbstractCoordinator(ConsumerNetworkClient client,
                               String groupId,
                               int rebalanceTimeoutMs,
                               int sessionTimeoutMs,
                               int heartbeatIntervalMs,
                               Metrics metrics,
                               String metricGrpPrefix,
                               Time time,
                               long retryBackoffMs,
                               boolean leaveGroupOnClose); }
AbstractCoordinator implements Closeable { protected synchronized RequestFuture<Void> lookupCoordinator() { if (findCoordinatorFuture == null) { Node node = this.client.leastLoadedNode(); if (node == null) { log.debug("No broker available to send GroupCoordinator request for group {}", groupId); return RequestFuture.noBrokersAvailable(); } else findCoordinatorFuture = sendGroupCoordinatorRequest(node); } return findCoordinatorFuture; } 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 { protected synchronized RequestFuture<Void> lookupCoordinator() { if (findCoordinatorFuture == null) { Node node = this.client.leastLoadedNode(); if (node == null) { log.debug("No broker available to send GroupCoordinator request for group {}", groupId); return RequestFuture.noBrokersAvailable(); } else findCoordinatorFuture = sendGroupCoordinatorRequest(node); } return findCoordinatorFuture; } 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();  }