test_class.declarator: topicName = "test"
test_class.modifier: private
test_class.original_string: private String topicName = "test";
test_class.type: String
test_class.var_name: topicName
test_class.declarator: groupId = "test-group"
test_class.modifier: private
test_class.original_string: private String groupId = "test-group";
test_class.type: String
test_class.var_name: groupId
test_class.declarator: tp = new TopicPartition(topicName, 0)
test_class.modifier: private
test_class.original_string: private TopicPartition tp = new TopicPartition(topicName, 0);
test_class.type: TopicPartition
test_class.var_name: tp
test_class.declarator: sessionTimeoutMs = 10000
test_class.modifier: private
test_class.original_string: private int sessionTimeoutMs = 10000;
test_class.type: int
test_class.var_name: sessionTimeoutMs
test_class.declarator: heartbeatIntervalMs = 5000
test_class.modifier: private
test_class.original_string: private int heartbeatIntervalMs = 5000;
test_class.type: int
test_class.var_name: heartbeatIntervalMs
test_class.declarator: retryBackoffMs = 100
test_class.modifier: private
test_class.original_string: private long retryBackoffMs = 100;
test_class.type: long
test_class.var_name: retryBackoffMs
test_class.declarator: autoCommitEnabled = false
test_class.modifier: private
test_class.original_string: private boolean autoCommitEnabled = false;
test_class.type: boolean
test_class.var_name: autoCommitEnabled
test_class.declarator: autoCommitIntervalMs = 2000
test_class.modifier: private
test_class.original_string: private long autoCommitIntervalMs = 2000;
test_class.type: long
test_class.var_name: autoCommitIntervalMs
test_class.declarator: partitionAssignor = new MockPartitionAssignor()
test_class.modifier: private
test_class.original_string: private MockPartitionAssignor partitionAssignor = new MockPartitionAssignor();
test_class.type: MockPartitionAssignor
test_class.var_name: partitionAssignor
test_class.declarator: assignors = Collections.<PartitionAssignor>singletonList(partitionAssignor)
test_class.modifier: private
test_class.original_string: private List<PartitionAssignor> assignors = Collections.<PartitionAssignor>singletonList(partitionAssignor);
test_class.type: List<PartitionAssignor>
test_class.var_name: assignors
test_class.declarator: time
test_class.modifier: private
test_class.original_string: private MockTime time;
test_class.type: MockTime
test_class.var_name: time
test_class.declarator: client
test_class.modifier: private
test_class.original_string: private MockClient client;
test_class.type: MockClient
test_class.var_name: client
test_class.declarator: cluster = TestUtils.singletonCluster(topicName, 1)
test_class.modifier: private
test_class.original_string: private Cluster cluster = TestUtils.singletonCluster(topicName, 1);
test_class.type: Cluster
test_class.var_name: cluster
test_class.declarator: node = cluster.nodes().get(0)
test_class.modifier: private
test_class.original_string: private Node node = cluster.nodes().get(0);
test_class.type: Node
test_class.var_name: node
test_class.declarator: subscriptions
test_class.modifier: private
test_class.original_string: private SubscriptionState subscriptions;
test_class.type: SubscriptionState
test_class.var_name: subscriptions
test_class.declarator: metadata
test_class.modifier: private
test_class.original_string: private Metadata metadata;
test_class.type: Metadata
test_class.var_name: metadata
test_class.declarator: metrics
test_class.modifier: private
test_class.original_string: private Metrics metrics;
test_class.type: Metrics
test_class.var_name: metrics
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: rebalanceListener
test_class.modifier: private
test_class.original_string: private MockRebalanceListener rebalanceListener;
test_class.type: MockRebalanceListener
test_class.var_name: rebalanceListener
test_class.declarator: defaultOffsetCommitCallback
test_class.modifier: private
test_class.original_string: private MockCommitCallback defaultOffsetCommitCallback;
test_class.type: MockCommitCallback
test_class.var_name: defaultOffsetCommitCallback
test_class.declarator: coordinator
test_class.modifier: private
test_class.original_string: private ConsumerCoordinator coordinator;
test_class.type: ConsumerCoordinator
test_class.var_name: coordinator
test_class.file: clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
test_class.identifier: ConsumerCoordinatorTest
test_case.body: @Test
      public void testMaybeLeaveGroup() {
          final String consumerId = "consumer";
  
          subscriptions.subscribe(Arrays.asList(topicName), rebalanceListener);
          subscriptions.needReassignment();
  
          client.prepareResponse(consumerMetadataResponse(node, Errors.NONE.code()));
          coordinator.ensureCoordinatorReady();
  
          client.prepareResponse(joinGroupFollowerResponse(1, consumerId, "leader", Errors.NONE.code()));
          client.prepareResponse(syncGroupResponse(Arrays.asList(tp), Errors.NONE.code()));
          coordinator.ensurePartitionAssignment();
  
          final AtomicBoolean received = new AtomicBoolean(false);
          client.prepareResponse(new MockClient.RequestMatcher() {
              @Override
              public boolean matches(ClientRequest request) {
                  received.set(true);
                  LeaveGroupRequest leaveRequest = new LeaveGroupRequest(request.request().body());
                  return leaveRequest.memberId().equals(consumerId) &&
                          leaveRequest.groupId().equals(groupId);
              }
          }, new LeaveGroupResponse(Errors.NONE.code()).toStruct());
          coordinator.maybeLeaveGroup();
          assertTrue(received.get());
          assertEquals(JoinGroupRequest.UNKNOWN_MEMBER_ID, coordinator.memberId);
          assertEquals(OffsetCommitRequest.DEFAULT_GENERATION_ID, coordinator.generation);
      }
test_case.class_method_signature: ConsumerCoordinatorTest.testMaybeLeaveGroup()
test_case.full_signature: @Test public void testMaybeLeaveGroup()
test_case.identifier: testMaybeLeaveGroup
test_case.invocations: subscribe asList needReassignment prepareResponse consumerMetadataResponse code ensureCoordinatorReady prepareResponse joinGroupFollowerResponse code prepareResponse syncGroupResponse asList code ensurePartitionAssignment prepareResponse set body request equals memberId equals groupId toStruct code maybeLeaveGroup assertTrue get assertEquals assertEquals
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testMaybeLeaveGroup()
focal_class.declarator: log = LoggerFactory.getLogger(ConsumerCoordinator.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger log = LoggerFactory.getLogger(ConsumerCoordinator.class);
focal_class.type: Logger
focal_class.var_name: log
focal_class.declarator: assignors
focal_class.modifier: private final
focal_class.original_string: private final List<PartitionAssignor> assignors;
focal_class.type: List<PartitionAssignor>
focal_class.var_name: assignors
focal_class.declarator: metadata
focal_class.modifier: private final
focal_class.original_string: private final Metadata metadata;
focal_class.type: Metadata
focal_class.var_name: metadata
focal_class.declarator: sensors
focal_class.modifier: private final
focal_class.original_string: private final ConsumerCoordinatorMetrics sensors;
focal_class.type: ConsumerCoordinatorMetrics
focal_class.var_name: sensors
focal_class.declarator: subscriptions
focal_class.modifier: private final
focal_class.original_string: private final SubscriptionState subscriptions;
focal_class.type: SubscriptionState
focal_class.var_name: subscriptions
focal_class.declarator: defaultOffsetCommitCallback
focal_class.modifier: private final
focal_class.original_string: private final OffsetCommitCallback defaultOffsetCommitCallback;
focal_class.type: OffsetCommitCallback
focal_class.var_name: defaultOffsetCommitCallback
focal_class.declarator: autoCommitEnabled
focal_class.modifier: private final
focal_class.original_string: private final boolean autoCommitEnabled;
focal_class.type: boolean
focal_class.var_name: autoCommitEnabled
focal_class.declarator: autoCommitTask
focal_class.modifier: private final
focal_class.original_string: private final AutoCommitTask autoCommitTask;
focal_class.type: AutoCommitTask
focal_class.var_name: autoCommitTask
focal_class.declarator: interceptors
focal_class.modifier: private final
focal_class.original_string: private final ConsumerInterceptors<?, ?> interceptors;
focal_class.type: ConsumerInterceptors<?, ?>
focal_class.var_name: interceptors
focal_class.declarator: excludeInternalTopics
focal_class.modifier: private final
focal_class.original_string: private final boolean excludeInternalTopics;
focal_class.type: boolean
focal_class.var_name: excludeInternalTopics
focal_class.declarator: metadataSnapshot
focal_class.modifier: private
focal_class.original_string: private MetadataSnapshot metadataSnapshot;
focal_class.type: MetadataSnapshot
focal_class.var_name: metadataSnapshot
focal_class.declarator: assignmentSnapshot
focal_class.modifier: private
focal_class.original_string: private MetadataSnapshot assignmentSnapshot;
focal_class.type: MetadataSnapshot
focal_class.var_name: assignmentSnapshot
focal_class.file: clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
focal_class.identifier: ConsumerCoordinator
focal_class.class_method_signature: ConsumerCoordinator.ConsumerCoordinator(ConsumerNetworkClient client,
          String groupId,
          int sessionTimeoutMs,
          int heartbeatIntervalMs,
          List<PartitionAssignor> assignors,
          Metadata metadata,
          SubscriptionState subscriptions,
          Metrics metrics,
          String metricGrpPrefix,
          Time time,
          long retryBackoffMs,
          OffsetCommitCallback defaultOffsetCommitCallback,
          boolean autoCommitEnabled,
          long autoCommitIntervalMs,
          ConsumerInterceptors<?, ?> interceptors,
          boolean excludeInternalTopics)
focal_class.full_signature: public  ConsumerCoordinator(ConsumerNetworkClient client,
          String groupId,
          int sessionTimeoutMs,
          int heartbeatIntervalMs,
          List<PartitionAssignor> assignors,
          Metadata metadata,
          SubscriptionState subscriptions,
          Metrics metrics,
          String metricGrpPrefix,
          Time time,
          long retryBackoffMs,
          OffsetCommitCallback defaultOffsetCommitCallback,
          boolean autoCommitEnabled,
          long autoCommitIntervalMs,
          ConsumerInterceptors<?, ?> interceptors,
          boolean excludeInternalTopics)
focal_class.identifier: ConsumerCoordinator
focal_class.modifiers: public
focal_class.parameters: (ConsumerNetworkClient client,
          String groupId,
          int sessionTimeoutMs,
          int heartbeatIntervalMs,
          List<PartitionAssignor> assignors,
          Metadata metadata,
          SubscriptionState subscriptions,
          Metrics metrics,
          String metricGrpPrefix,
          Time time,
          long retryBackoffMs,
          OffsetCommitCallback defaultOffsetCommitCallback,
          boolean autoCommitEnabled,
          long autoCommitIntervalMs,
          ConsumerInterceptors<?, ?> interceptors,
          boolean excludeInternalTopics)
focal_class.signature:  ConsumerCoordinator(ConsumerNetworkClient client,
          String groupId,
          int sessionTimeoutMs,
          int heartbeatIntervalMs,
          List<PartitionAssignor> assignors,
          Metadata metadata,
          SubscriptionState subscriptions,
          Metrics metrics,
          String metricGrpPrefix,
          Time time,
          long retryBackoffMs,
          OffsetCommitCallback defaultOffsetCommitCallback,
          boolean autoCommitEnabled,
          long autoCommitIntervalMs,
          ConsumerInterceptors<?, ?> interceptors,
          boolean excludeInternalTopics)
focal_class.class_method_signature: ConsumerCoordinator.protocolType()
focal_class.full_signature: @Override public String protocolType()
focal_class.identifier: protocolType
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: String
focal_class.signature: String protocolType()
focal_class.class_method_signature: ConsumerCoordinator.metadata()
focal_class.full_signature: @Override public List<ProtocolMetadata> metadata()
focal_class.identifier: metadata
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: List<ProtocolMetadata>
focal_class.signature: List<ProtocolMetadata> metadata()
focal_class.class_method_signature: ConsumerCoordinator.addMetadataListener()
focal_class.full_signature: private void addMetadataListener()
focal_class.identifier: addMetadataListener
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void addMetadataListener()
focal_class.class_method_signature: ConsumerCoordinator.filterTopic(String topic)
focal_class.full_signature: private boolean filterTopic(String topic)
focal_class.identifier: filterTopic
focal_class.modifiers: private
focal_class.parameters: (String topic)
focal_class.return: boolean
focal_class.signature: boolean filterTopic(String topic)
focal_class.class_method_signature: ConsumerCoordinator.lookupAssignor(String name)
focal_class.full_signature: private PartitionAssignor lookupAssignor(String name)
focal_class.identifier: lookupAssignor
focal_class.modifiers: private
focal_class.parameters: (String name)
focal_class.return: PartitionAssignor
focal_class.signature: PartitionAssignor lookupAssignor(String name)
focal_class.class_method_signature: ConsumerCoordinator.performAssignment(String leaderId,
          String assignmentStrategy,
          Map<String/* memberId */, ByteBuffer/* 包含它关注了那些 topic */> allSubscriptions) {
focal_class.full_signature: @Override protected Map<String, ByteBuffer> performAssignment(String leaderId,
          String assignmentStrategy,
          Map<String/* memberId */, ByteBuffer/* 包含它关注了那些 topic */> allSubscriptions) {
focal_class.identifier: performAssignment
focal_class.modifiers: @Override protected
focal_class.parameters: (String leaderId,
          String assignmentStrategy,
          Map<String/* memberId */, ByteBuffer/* 包含它关注了那些 topic */> allSubscriptions) {
focal_class.return: Map<String, ByteBuffer>
focal_class.signature: Map<String, ByteBuffer> performAssignment(String leaderId,
          String assignmentStrategy,
          Map<String/* memberId */, ByteBuffer/* 包含它关注了那些 topic */> allSubscriptions) {
focal_class.class_method_signature: ConsumerCoordinator.onJoinComplete(int generation,
          String memberId,
          String assignmentStrategy,
          ByteBuffer assignmentBuffer)
focal_class.full_signature: @Override protected void onJoinComplete(int generation,
          String memberId,
          String assignmentStrategy,
          ByteBuffer assignmentBuffer)
focal_class.identifier: onJoinComplete
focal_class.modifiers: @Override protected
focal_class.parameters: (int generation,
          String memberId,
          String assignmentStrategy,
          ByteBuffer assignmentBuffer)
focal_class.return: void
focal_class.signature: void onJoinComplete(int generation,
          String memberId,
          String assignmentStrategy,
          ByteBuffer assignmentBuffer)
focal_class.class_method_signature: ConsumerCoordinator.onJoinPrepare(int generation, String memberId)
focal_class.full_signature: @Override protected void onJoinPrepare(int generation, String memberId)
focal_class.identifier: onJoinPrepare
focal_class.modifiers: @Override protected
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: ConsumerCoordinator.needRejoin()
focal_class.full_signature: @Override public boolean needRejoin()
focal_class.identifier: needRejoin
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: boolean
focal_class.signature: boolean needRejoin()
focal_class.class_method_signature: ConsumerCoordinator.refreshCommittedOffsetsIfNeeded()
focal_class.full_signature: public void refreshCommittedOffsetsIfNeeded()
focal_class.identifier: refreshCommittedOffsetsIfNeeded
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void refreshCommittedOffsetsIfNeeded()
focal_class.class_method_signature: ConsumerCoordinator.fetchCommittedOffsets(Set<TopicPartition> partitions)
focal_class.full_signature: public Map<TopicPartition, OffsetAndMetadata> fetchCommittedOffsets(Set<TopicPartition> partitions)
focal_class.identifier: fetchCommittedOffsets
focal_class.modifiers: public
focal_class.parameters: (Set<TopicPartition> partitions)
focal_class.return: Map<TopicPartition, OffsetAndMetadata>
focal_class.signature: Map<TopicPartition, OffsetAndMetadata> fetchCommittedOffsets(Set<TopicPartition> partitions)
focal_class.class_method_signature: ConsumerCoordinator.ensurePartitionAssignment()
focal_class.full_signature: public void ensurePartitionAssignment()
focal_class.identifier: ensurePartitionAssignment
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void ensurePartitionAssignment()
focal_class.class_method_signature: ConsumerCoordinator.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: ConsumerCoordinator.commitOffsetsAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback)
focal_class.full_signature: public void commitOffsetsAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback)
focal_class.identifier: commitOffsetsAsync
focal_class.modifiers: public
focal_class.parameters: (final Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback)
focal_class.return: void
focal_class.signature: void commitOffsetsAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback)
focal_class.class_method_signature: ConsumerCoordinator.commitOffsetsSync(Map<TopicPartition, OffsetAndMetadata> offsets)
focal_class.full_signature: public void commitOffsetsSync(Map<TopicPartition, OffsetAndMetadata> offsets)
focal_class.identifier: commitOffsetsSync
focal_class.modifiers: public
focal_class.parameters: (Map<TopicPartition, OffsetAndMetadata> offsets)
focal_class.return: void
focal_class.signature: void commitOffsetsSync(Map<TopicPartition, OffsetAndMetadata> offsets)
focal_class.class_method_signature: ConsumerCoordinator.maybeAutoCommitOffsetsSync()
focal_class.full_signature: private void maybeAutoCommitOffsetsSync()
focal_class.identifier: maybeAutoCommitOffsetsSync
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void maybeAutoCommitOffsetsSync()
focal_class.class_method_signature: ConsumerCoordinator.sendOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets)
focal_class.full_signature: private RequestFuture<Void> sendOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets)
focal_class.identifier: sendOffsetCommitRequest
focal_class.modifiers: private
focal_class.parameters: (final Map<TopicPartition, OffsetAndMetadata> offsets)
focal_class.return: RequestFuture<Void>
focal_class.signature: RequestFuture<Void> sendOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets)
focal_class.class_method_signature: ConsumerCoordinator.sendOffsetFetchRequest(Set<TopicPartition> partitions)
focal_class.full_signature: private RequestFuture<Map<TopicPartition, OffsetAndMetadata>> sendOffsetFetchRequest(Set<TopicPartition> partitions)
focal_class.identifier: sendOffsetFetchRequest
focal_class.modifiers: private
focal_class.parameters: (Set<TopicPartition> partitions)
focal_class.return: RequestFuture<Map<TopicPartition, OffsetAndMetadata>>
focal_class.signature: RequestFuture<Map<TopicPartition, OffsetAndMetadata>> sendOffsetFetchRequest(Set<TopicPartition> partitions)
focal_class.superclass: extends AbstractCoordinator
focal_method.body: public void ensurePartitionAssignment() {
          if (subscriptions.partitionsAutoAssigned()) {// SubscriptionType.AUTO_TOPICS 或 SubscriptionType.AUTO_PATTERN;
  
              // Due to a race condition between the initial metadata fetch and the initial rebalance, we need to ensure that
              // the metadata is fresh before joining initially, and then request the metadata update. If metadata update arrives
              // while the rebalance is still pending (for example, when the join group is still inflight), then we will lose
              // track of the fact that we need to rebalance again to reflect the change to the topic subscription. Without
              // ensuring that the metadata is fresh, any metadata update that changes the topic subscriptions and arrives with a
              // rebalance in progress will essentially be ignored. See KAFKA-3949 for the complete description of the problem.
  
              // 基于初始化获取metadata与初始化Rebalance的竞态条件，我们需要确保在进行初始化时，metadata是新的，然后
              // 请求metadata的更新。如果在rebalance还在等待时metadata的更新抵达了（比如，加入group的操作依旧inflight），
              // 我们将会丢失真实的踪迹？，所以我们需要再次rebalance来反应topic订阅情况的改变。在没有确保metadata是新鲜的之前，
              // 任何改变topic订阅以及在rebalance时到来的metadata更新将会被忽略。完整表述这个问题，可以看看 KAFKA-3949
  
              if (subscriptions.hasPatternSubscription()) {// SubscriptionType.AUTO_PATTERN
                  client.ensureFreshMetadata();// 防止因为使用过期的Metadata进行Rebalance而导致多次连续的Rebalance
              }
  
              ensureActiveGroup();
          }
      }
focal_method.class_method_signature: ConsumerCoordinator.ensurePartitionAssignment()
focal_method.full_signature: public void ensurePartitionAssignment()
focal_method.identifier: ensurePartitionAssignment
focal_method.invocations: partitionsAutoAssigned hasPatternSubscription ensureFreshMetadata ensureActiveGroup
focal_method.modifiers: public
focal_method.parameters: ()
focal_method.return: void
focal_method.signature: void ensurePartitionAssignment()