test_class.declarator: EMPTY_DATA = ByteBuffer.wrap(new byte[0])
test_class.modifier: private static final
test_class.original_string: private static final ByteBuffer EMPTY_DATA = ByteBuffer.wrap(new byte[0]);
test_class.type: ByteBuffer
test_class.var_name: EMPTY_DATA
test_class.declarator: SESSION_TIMEOUT_MS = 30000
test_class.modifier: private static final
test_class.original_string: private static final int SESSION_TIMEOUT_MS = 30000;
test_class.type: int
test_class.var_name: SESSION_TIMEOUT_MS
test_class.declarator: HEARTBEAT_INTERVAL_MS = 3000
test_class.modifier: private static final
test_class.original_string: private static final int HEARTBEAT_INTERVAL_MS = 3000;
test_class.type: int
test_class.var_name: HEARTBEAT_INTERVAL_MS
test_class.declarator: RETRY_BACKOFF_MS = 100
test_class.modifier: private static final
test_class.original_string: private static final long RETRY_BACKOFF_MS = 100;
test_class.type: long
test_class.var_name: RETRY_BACKOFF_MS
test_class.declarator: REQUEST_TIMEOUT_MS = 40000
test_class.modifier: private static final
test_class.original_string: private static final long REQUEST_TIMEOUT_MS = 40000;
test_class.type: long
test_class.var_name: REQUEST_TIMEOUT_MS
test_class.declarator: GROUP_ID = "dummy-group"
test_class.modifier: private static final
test_class.original_string: private static final String GROUP_ID = "dummy-group";
test_class.type: String
test_class.var_name: GROUP_ID
test_class.declarator: METRIC_GROUP_PREFIX = "consumer"
test_class.modifier: private static final
test_class.original_string: private static final String METRIC_GROUP_PREFIX = "consumer";
test_class.type: String
test_class.var_name: METRIC_GROUP_PREFIX
test_class.declarator: mockClient
test_class.modifier: private
test_class.original_string: private MockClient mockClient;
test_class.type: MockClient
test_class.var_name: mockClient
test_class.declarator: mockTime
test_class.modifier: private
test_class.original_string: private MockTime mockTime;
test_class.type: MockTime
test_class.var_name: mockTime
test_class.declarator: node
test_class.modifier: private
test_class.original_string: private Node node;
test_class.type: Node
test_class.var_name: node
test_class.declarator: coordinatorNode
test_class.modifier: private
test_class.original_string: private Node coordinatorNode;
test_class.type: Node
test_class.var_name: coordinatorNode
test_class.declarator: consumerClient
test_class.modifier: private
test_class.original_string: private ConsumerNetworkClient consumerClient;
test_class.type: ConsumerNetworkClient
test_class.var_name: consumerClient
test_class.declarator: coordinator
test_class.modifier: private
test_class.original_string: private DummyCoordinator coordinator;
test_class.type: DummyCoordinator
test_class.var_name: coordinator
test_class.file: clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
test_class.identifier: AbstractCoordinatorTest
test_case.body: @Test
      public void testCoordinatorDiscoveryBackoff() {
          mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code()));
          mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code()));
  
          // blackout the coordinator for 50 milliseconds to simulate a disconnect.
          // after backing off, we should be able to connect.
          mockClient.blackout(coordinatorNode, 50L);
  
          long initialTime = mockTime.milliseconds();
          coordinator.ensureCoordinatorReady();
          long endTime = mockTime.milliseconds();
  
          assertTrue(endTime - initialTime >= RETRY_BACKOFF_MS);
      }
test_case.class_method_signature: AbstractCoordinatorTest.testCoordinatorDiscoveryBackoff()
test_case.full_signature: @Test public void testCoordinatorDiscoveryBackoff()
test_case.identifier: testCoordinatorDiscoveryBackoff
test_case.invocations: prepareResponse groupCoordinatorResponse code prepareResponse groupCoordinatorResponse code blackout milliseconds ensureCoordinatorReady milliseconds assertTrue
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testCoordinatorDiscoveryBackoff()
focal_class.declarator: log = LoggerFactory.getLogger(AbstractCoordinator.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger log = LoggerFactory.getLogger(AbstractCoordinator.class);
focal_class.type: Logger
focal_class.var_name: log
focal_class.declarator: heartbeat
focal_class.modifier: private final
focal_class.original_string: private final Heartbeat heartbeat;
focal_class.type: Heartbeat
focal_class.var_name: heartbeat
focal_class.declarator: heartbeatTask
focal_class.modifier: private final
focal_class.original_string: private final HeartbeatTask heartbeatTask;
focal_class.type: HeartbeatTask
focal_class.var_name: heartbeatTask
focal_class.declarator: sessionTimeoutMs
focal_class.modifier: private final
focal_class.original_string: private final int sessionTimeoutMs;
focal_class.type: int
focal_class.var_name: sessionTimeoutMs
focal_class.declarator: sensors
focal_class.modifier: private final
focal_class.original_string: private final GroupCoordinatorMetrics sensors;
focal_class.type: GroupCoordinatorMetrics
focal_class.var_name: sensors
focal_class.declarator: groupId
focal_class.modifier: protected final
focal_class.original_string: protected final String groupId;
focal_class.type: String
focal_class.var_name: groupId
focal_class.declarator: client
focal_class.modifier: protected final
focal_class.original_string: protected final ConsumerNetworkClient client;
focal_class.type: ConsumerNetworkClient
focal_class.var_name: client
focal_class.declarator: time
focal_class.modifier: protected final
focal_class.original_string: protected final Time time;
focal_class.type: Time
focal_class.var_name: time
focal_class.declarator: retryBackoffMs
focal_class.modifier: protected final
focal_class.original_string: protected final long retryBackoffMs;
focal_class.type: long
focal_class.var_name: retryBackoffMs
focal_class.declarator: needsJoinPrepare = true
focal_class.modifier: private
focal_class.original_string: private boolean needsJoinPrepare = true;
focal_class.type: boolean
focal_class.var_name: needsJoinPrepare
focal_class.declarator: rejoinNeeded = true
focal_class.modifier: private
focal_class.original_string: private boolean rejoinNeeded = true;
focal_class.type: boolean
focal_class.var_name: rejoinNeeded
focal_class.declarator: coordinator
focal_class.modifier: protected
focal_class.original_string: protected Node coordinator;
focal_class.type: Node
focal_class.var_name: coordinator
focal_class.declarator: memberId
focal_class.modifier: protected
focal_class.original_string: protected String memberId;
focal_class.type: String
focal_class.var_name: memberId
focal_class.declarator: protocol
focal_class.modifier: protected
focal_class.original_string: protected String protocol;
focal_class.type: String
focal_class.var_name: protocol
focal_class.declarator: generation
focal_class.modifier: protected
focal_class.original_string: protected int generation;
focal_class.type: int
focal_class.var_name: generation
focal_class.file: clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
focal_class.identifier: AbstractCoordinator
focal_class.interfaces: implements Closeable
focal_class.class_method_signature: AbstractCoordinator.AbstractCoordinator(ConsumerNetworkClient client,
          String groupId,
          int sessionTimeoutMs,
          int heartbeatIntervalMs,
          Metrics metrics,
          String metricGrpPrefix,
          Time time,
          long retryBackoffMs)
focal_class.full_signature: public  AbstractCoordinator(ConsumerNetworkClient client,
          String groupId,
          int sessionTimeoutMs,
          int heartbeatIntervalMs,
          Metrics metrics,
          String metricGrpPrefix,
          Time time,
          long retryBackoffMs)
focal_class.identifier: AbstractCoordinator
focal_class.modifiers: public
focal_class.parameters: (ConsumerNetworkClient client,
          String groupId,
          int sessionTimeoutMs,
          int heartbeatIntervalMs,
          Metrics metrics,
          String metricGrpPrefix,
          Time time,
          long retryBackoffMs)
focal_class.signature:  AbstractCoordinator(ConsumerNetworkClient client,
          String groupId,
          int sessionTimeoutMs,
          int heartbeatIntervalMs,
          Metrics metrics,
          String metricGrpPrefix,
          Time time,
          long retryBackoffMs)
focal_class.class_method_signature: AbstractCoordinator.protocolType()
focal_class.full_signature: protected abstract String protocolType()
focal_class.identifier: protocolType
focal_class.modifiers: protected abstract
focal_class.parameters: ()
focal_class.return: String
focal_class.signature: String protocolType()
focal_class.class_method_signature: AbstractCoordinator.metadata()
focal_class.full_signature: protected abstract List<ProtocolMetadata> metadata()
focal_class.identifier: metadata
focal_class.modifiers: protected abstract
focal_class.parameters: ()
focal_class.return: List<ProtocolMetadata>
focal_class.signature: List<ProtocolMetadata> metadata()
focal_class.class_method_signature: AbstractCoordinator.onJoinPrepare(int generation, String memberId)
focal_class.full_signature: protected abstract void onJoinPrepare(int generation, String memberId)
focal_class.identifier: onJoinPrepare
focal_class.modifiers: protected abstract
focal_class.parameters: (int generation, String memberId)
focal_class.return: void
focal_class.signature: void onJoinPrepare(int generation, String memberId)
focal_class.class_method_signature: AbstractCoordinator.performAssignment(String leaderId,
          String protocol,
          Map<String, ByteBuffer> allMemberMetadata)
focal_class.full_signature: protected abstract Map<String, ByteBuffer> performAssignment(String leaderId,
          String protocol,
          Map<String, ByteBuffer> allMemberMetadata)
focal_class.identifier: performAssignment
focal_class.modifiers: protected abstract
focal_class.parameters: (String leaderId,
          String protocol,
          Map<String, ByteBuffer> allMemberMetadata)
focal_class.return: Map<String, ByteBuffer>
focal_class.signature: Map<String, ByteBuffer> performAssignment(String leaderId,
          String protocol,
          Map<String, ByteBuffer> allMemberMetadata)
focal_class.class_method_signature: AbstractCoordinator.onJoinComplete(int generation,
          String memberId,
          String protocol,
          ByteBuffer memberAssignment)
focal_class.full_signature: protected abstract void onJoinComplete(int generation,
          String memberId,
          String protocol,
          ByteBuffer memberAssignment)
focal_class.identifier: onJoinComplete
focal_class.modifiers: protected abstract
focal_class.parameters: (int generation,
          String memberId,
          String protocol,
          ByteBuffer memberAssignment)
focal_class.return: void
focal_class.signature: void onJoinComplete(int generation,
          String memberId,
          String protocol,
          ByteBuffer memberAssignment)
focal_class.class_method_signature: AbstractCoordinator.ensureCoordinatorReady()
focal_class.full_signature: public void ensureCoordinatorReady()
focal_class.identifier: ensureCoordinatorReady
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void ensureCoordinatorReady()
focal_class.class_method_signature: AbstractCoordinator.needRejoin()
focal_class.full_signature: protected boolean needRejoin()
focal_class.identifier: needRejoin
focal_class.modifiers: protected
focal_class.parameters: ()
focal_class.return: boolean
focal_class.signature: boolean needRejoin()
focal_class.class_method_signature: AbstractCoordinator.ensureActiveGroup()
focal_class.full_signature: public void ensureActiveGroup()
focal_class.identifier: ensureActiveGroup
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void ensureActiveGroup()
focal_class.class_method_signature: AbstractCoordinator.sendJoinGroupRequest()
focal_class.full_signature: private RequestFuture<ByteBuffer> sendJoinGroupRequest()
focal_class.identifier: sendJoinGroupRequest
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: RequestFuture<ByteBuffer>
focal_class.signature: RequestFuture<ByteBuffer> sendJoinGroupRequest()
focal_class.class_method_signature: AbstractCoordinator.onJoinFollower()
focal_class.full_signature: private RequestFuture<ByteBuffer> onJoinFollower()
focal_class.identifier: onJoinFollower
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: RequestFuture<ByteBuffer>
focal_class.signature: RequestFuture<ByteBuffer> onJoinFollower()
focal_class.class_method_signature: AbstractCoordinator.onJoinLeader(JoinGroupResponse joinResponse)
focal_class.full_signature: private RequestFuture<ByteBuffer> onJoinLeader(JoinGroupResponse joinResponse)
focal_class.identifier: onJoinLeader
focal_class.modifiers: private
focal_class.parameters: (JoinGroupResponse joinResponse)
focal_class.return: RequestFuture<ByteBuffer>
focal_class.signature: RequestFuture<ByteBuffer> onJoinLeader(JoinGroupResponse joinResponse)
focal_class.class_method_signature: AbstractCoordinator.sendSyncGroupRequest(SyncGroupRequest request)
focal_class.full_signature: private RequestFuture<ByteBuffer> sendSyncGroupRequest(SyncGroupRequest request)
focal_class.identifier: sendSyncGroupRequest
focal_class.modifiers: private
focal_class.parameters: (SyncGroupRequest request)
focal_class.return: RequestFuture<ByteBuffer>
focal_class.signature: RequestFuture<ByteBuffer> sendSyncGroupRequest(SyncGroupRequest request)
focal_class.class_method_signature: AbstractCoordinator.sendGroupCoordinatorRequest()
focal_class.full_signature: private RequestFuture<Void> sendGroupCoordinatorRequest()
focal_class.identifier: sendGroupCoordinatorRequest
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: RequestFuture<Void>
focal_class.signature: RequestFuture<Void> sendGroupCoordinatorRequest()
focal_class.class_method_signature: AbstractCoordinator.handleGroupMetadataResponse(ClientResponse resp, RequestFuture<Void> future)
focal_class.full_signature: private void handleGroupMetadataResponse(ClientResponse resp, RequestFuture<Void> future)
focal_class.identifier: handleGroupMetadataResponse
focal_class.modifiers: private
focal_class.parameters: (ClientResponse resp, RequestFuture<Void> future)
focal_class.return: void
focal_class.signature: void handleGroupMetadataResponse(ClientResponse resp, RequestFuture<Void> future)
focal_class.class_method_signature: AbstractCoordinator.coordinatorUnknown()
focal_class.full_signature: public boolean coordinatorUnknown()
focal_class.identifier: coordinatorUnknown
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: boolean
focal_class.signature: boolean coordinatorUnknown()
focal_class.class_method_signature: AbstractCoordinator.coordinatorDead()
focal_class.full_signature: protected void coordinatorDead()
focal_class.identifier: coordinatorDead
focal_class.modifiers: protected
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void coordinatorDead()
focal_class.class_method_signature: AbstractCoordinator.close()
focal_class.full_signature: @Override public void close()
focal_class.identifier: close
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void close()
focal_class.class_method_signature: AbstractCoordinator.maybeLeaveGroup()
focal_class.full_signature: public void maybeLeaveGroup()
focal_class.identifier: maybeLeaveGroup
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void maybeLeaveGroup()
focal_class.class_method_signature: AbstractCoordinator.sendLeaveGroupRequest()
focal_class.full_signature: private void sendLeaveGroupRequest()
focal_class.identifier: sendLeaveGroupRequest
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void sendLeaveGroupRequest()
focal_class.class_method_signature: AbstractCoordinator.sendHeartbeatRequest()
focal_class.full_signature: public RequestFuture<Void> sendHeartbeatRequest()
focal_class.identifier: sendHeartbeatRequest
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: RequestFuture<Void>
focal_class.signature: RequestFuture<Void> sendHeartbeatRequest()
focal_method.body: public void ensureCoordinatorReady() {
          while (coordinatorUnknown()) {// 是否需要重新查找GroupCoordinator，主要检查coordinator字段是否为空，以及与GroupCoordinator
              // 之间的连接是否正常
              RequestFuture<Void> future = sendGroupCoordinatorRequest();// 创建并缓存请求
              client.poll(future);// 阻塞地发送请求
  
              if (future.failed()) {
                  if (future.isRetriable()) {
                      client.awaitMetadataUpdate();
                  } else {
                      throw future.exception();
                  }
              } else if (coordinator != null && client.connectionFailed(coordinator)) {
                  // 如果Coordinator 依旧为null，并且client连接Coordinator节点失败，这里sleep一段时间，然后继续查找Coordinator
                  // we found the coordinator, but the connection has failed, so mark
                  // it dead and backoff before retrying discovery
                  coordinatorDead();
                  time.sleep(retryBackoffMs);
              }
          }
      }
focal_method.class_method_signature: AbstractCoordinator.ensureCoordinatorReady()
focal_method.full_signature: public void ensureCoordinatorReady()
focal_method.identifier: ensureCoordinatorReady
focal_method.invocations: coordinatorUnknown sendGroupCoordinatorRequest poll failed isRetriable awaitMetadataUpdate exception connectionFailed coordinatorDead sleep
focal_method.modifiers: public
focal_method.parameters: ()
focal_method.return: void
focal_method.signature: void ensureCoordinatorReady()