content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements. See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License. You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.kafka.clients.consumer.internals;
  
  import org.apache.kafka.clients.ClientResponse;
  import org.apache.kafka.clients.consumer.internals.events.BackgroundEventHandler;
  import org.apache.kafka.common.Node;
  import org.apache.kafka.common.errors.TimeoutException;
  import org.apache.kafka.common.protocol.ApiKeys;
  import org.apache.kafka.common.protocol.Errors;
  import org.apache.kafka.common.requests.AbstractRequest;
  import org.apache.kafka.common.requests.FindCoordinatorRequest;
  import org.apache.kafka.common.requests.FindCoordinatorResponse;
  import org.apache.kafka.common.requests.RequestHeader;
  import org.apache.kafka.common.test.api.Flaky;
  import org.apache.kafka.common.utils.LogCaptureAppender;
  import org.apache.kafka.common.utils.LogContext;
  import org.apache.kafka.common.utils.MockTime;
  
  import org.apache.logging.log4j.Level;
  import org.junit.jupiter.api.BeforeEach;
  import org.junit.jupiter.api.Test;
  import org.junit.jupiter.params.ParameterizedTest;
  import org.junit.jupiter.params.provider.EnumSource;
  
  import java.util.Collections;
  import java.util.List;
  import java.util.Objects;
  import java.util.Optional;
  import java.util.regex.Matcher;
  import java.util.regex.Pattern;
  import java.util.stream.Collectors;
  
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertFalse;
  import static org.junit.jupiter.api.Assertions.assertInstanceOf;
  import static org.junit.jupiter.api.Assertions.assertThrows;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  import static org.mockito.Mockito.mock;
  import static org.mockito.Mockito.verifyNoInteractions;
  
  public class CoordinatorRequestManagerTest {
      private static final int RETRY_BACKOFF_MS = 500;
      private static final String GROUP_ID = "group-1";
      private MockTime time;
      private BackgroundEventHandler backgroundEventHandler;
      private Node node;
  
      @BeforeEach
      public void setup() {
          this.time = new MockTime(0);
          this.node = new Node(1, "localhost", 9092);
          this.backgroundEventHandler = mock(BackgroundEventHandler.class);
      }
  
      @Test
      public void testSuccessfulResponse() {
          CoordinatorRequestManager coordinatorManager = setupCoordinatorManager(GROUP_ID);
          expectFindCoordinatorRequest(coordinatorManager, Errors.NONE);
  
          Optional<Node> coordinatorOpt = coordinatorManager.coordinator();
          assertTrue(coordinatorOpt.isPresent());
          assertEquals(Integer.MAX_VALUE - node.id(), coordinatorOpt.get().id());
          assertEquals(node.host(), coordinatorOpt.get().host());
          assertEquals(node.port(), coordinatorOpt.get().port());
  
          NetworkClientDelegate.PollResult pollResult = coordinatorManager.poll(time.milliseconds());
          assertEquals(Collections.emptyList(), pollResult.unsentRequests);
      }
  
      /**
       * This test mimics a client that has been disconnected from the coordinator. When the client remains disconnected
       * from the coordinator for 60 seconds, the client will begin to emit a warning log every minute thereafter to
       * alert the user about the ongoing disconnect status. The warning log includes the length of time of the ongoing
       * disconnect:
       *
       * <code>
       *     Consumer has been disconnected from the group coordinator for XXXXXms
       * </code>
       *
       * <p/>
       *
       * However, the logic used to calculate the length of the disconnect was not correct. This test exercises the
       * disconnect logic, controlling the logging and system time, to ensure the warning message is correct.
       *
       * @see CoordinatorRequestManager#markCoordinatorUnknown(String, long)
       */
      @Flaky("KAFKA-18776")
      @Test
      public void testMarkCoordinatorUnknownLoggingAccuracy() {
          long oneMinute = 60000;
  
          try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister()) {
              appender.setClassLogger(CoordinatorRequestManager.class, Level.WARN);
              CoordinatorRequestManager coordinatorRequestManager = setupCoordinatorManager(GROUP_ID);
              assertFalse(coordinatorRequestManager.coordinator().isPresent());
  
              // Step 1: mark the coordinator as disconnected right after creation of the CoordinatorRequestManager.
              // Because the disconnect occurred immediately, no warning should be logged.
              coordinatorRequestManager.markCoordinatorUnknown("test", time.milliseconds());
              assertTrue(millisecondsFromLog(appender).isEmpty());
  
              // Step 2: sleep for one minute and mark the coordinator unknown again. Then verify that the warning was
              // logged and the reported time is accurate.
              time.sleep(oneMinute);
              coordinatorRequestManager.markCoordinatorUnknown("test", time.milliseconds());
              Optional<Long> firstLogMs = millisecondsFromLog(appender);
              assertTrue(firstLogMs.isPresent());
              assertEquals(oneMinute, firstLogMs.get());
  
              // Step 3: sleep for *another* minute, mark the coordinator unknown again, and verify the accuracy.
              time.sleep(oneMinute);
              coordinatorRequestManager.markCoordinatorUnknown("test", time.milliseconds());
              Optional<Long> secondLogMs = millisecondsFromLog(appender);
              assertTrue(secondLogMs.isPresent());
              assertEquals(oneMinute * 2, secondLogMs.get());
          }
      }
  
      private Optional<Long> millisecondsFromLog(LogCaptureAppender appender) {
          Pattern pattern = Pattern.compile("^Consumer has been disconnected from the group coordinator for (?<millis>\\d+)+ms$");
          List<Long> milliseconds = appender.getMessages().stream()
              .map(pattern::matcher)
              .filter(Matcher::find)
              .map(matcher -> matcher.group("millis"))
              .filter(Objects::nonNull)
              .map(millisString -> {
                  try {
                      return Long.parseLong(millisString);
                  } catch (NumberFormatException e) {
                      return null;
                  }
              })
              .filter(Objects::nonNull)
              .collect(Collectors.toList());
  
          // Return the most recent log entry that matches the message in markCoordinatorUnknown, if present.
          return milliseconds.isEmpty() ? Optional.empty() : Optional.of(milliseconds.get(milliseconds.size() - 1));
      }
  
      @Test
      public void testMarkCoordinatorUnknown() {
          CoordinatorRequestManager coordinatorManager = setupCoordinatorManager(GROUP_ID);
  
          expectFindCoordinatorRequest(coordinatorManager, Errors.NONE);
          assertTrue(coordinatorManager.coordinator().isPresent());
  
          // It may take time for metadata to converge between after a coordinator has
          // been demoted. This can cause a tight loop in which FindCoordinator continues to
          // return node X while that node continues to reply with NOT_COORDINATOR. Hence we
          // still want to ensure a backoff after successfully finding the coordinator.
          coordinatorManager.markCoordinatorUnknown("coordinator changed", time.milliseconds());
          assertEquals(Collections.emptyList(), coordinatorManager.poll(time.milliseconds()).unsentRequests);
  
          time.sleep(RETRY_BACKOFF_MS - 1);
          assertEquals(Collections.emptyList(), coordinatorManager.poll(time.milliseconds()).unsentRequests);
  
          time.sleep(RETRY_BACKOFF_MS);
          expectFindCoordinatorRequest(coordinatorManager, Errors.NONE);
          assertTrue(coordinatorManager.coordinator().isPresent());
      }
  
      @Test
      public void testBackoffAfterRetriableFailure() {
          CoordinatorRequestManager coordinatorManager = setupCoordinatorManager(GROUP_ID);
          expectFindCoordinatorRequest(coordinatorManager, Errors.COORDINATOR_LOAD_IN_PROGRESS);
          verifyNoInteractions(backgroundEventHandler);
  
          time.sleep(RETRY_BACKOFF_MS - 1);
          assertEquals(Collections.emptyList(), coordinatorManager.poll(time.milliseconds()).unsentRequests);
  
          time.sleep(1);
          expectFindCoordinatorRequest(coordinatorManager, Errors.NONE);
      }
  
      @Test
      public void testBackoffAfterFatalError() {
          CoordinatorRequestManager coordinatorManager = setupCoordinatorManager(GROUP_ID);
          expectFindCoordinatorRequest(coordinatorManager, Errors.GROUP_AUTHORIZATION_FAILED);
  
          time.sleep(RETRY_BACKOFF_MS - 1);
          assertEquals(Collections.emptyList(), coordinatorManager.poll(time.milliseconds()).unsentRequests);
  
          time.sleep(1);
          assertEquals(1, coordinatorManager.poll(time.milliseconds()).unsentRequests.size());
          assertEquals(Optional.empty(), coordinatorManager.coordinator());
      }
  
      @Test
      public void testNullGroupIdShouldThrow() {
          assertThrows(RuntimeException.class, () -> setupCoordinatorManager(null));
      }
  
      @Test
      public void testFindCoordinatorResponseVersions() {
          // v4
          FindCoordinatorResponse respNew = FindCoordinatorResponse.prepareResponse(Errors.NONE, GROUP_ID, this.node);
          assertTrue(respNew.coordinatorByKey(GROUP_ID).isPresent());
          assertEquals(GROUP_ID, respNew.coordinatorByKey(GROUP_ID).get().key());
          assertEquals(this.node.id(), respNew.coordinatorByKey(GROUP_ID).get().nodeId());
  
          // <= v3
          FindCoordinatorResponse respOld = FindCoordinatorResponse.prepareOldResponse(Errors.NONE, this.node);
          assertTrue(respOld.coordinatorByKey(GROUP_ID).isPresent());
          assertEquals(this.node.id(), respNew.coordinatorByKey(GROUP_ID).get().nodeId());
      }
  
      @Test
      public void testNetworkTimeout() {
          CoordinatorRequestManager coordinatorManager = setupCoordinatorManager(GROUP_ID);
          NetworkClientDelegate.PollResult res = coordinatorManager.poll(time.milliseconds());
          assertEquals(1, res.unsentRequests.size());
  
          // Mimic a network timeout
          res.unsentRequests.get(0).handler().onFailure(time.milliseconds(), new TimeoutException());
  
          // Sleep for exponential backoff - 1ms
          time.sleep(RETRY_BACKOFF_MS - 1);
          NetworkClientDelegate.PollResult res2 = coordinatorManager.poll(this.time.milliseconds());
          assertEquals(0, res2.unsentRequests.size());
  
          time.sleep(1);
          res2 = coordinatorManager.poll(time.milliseconds());
          assertEquals(1, res2.unsentRequests.size());
      }
  
      @ParameterizedTest
      @EnumSource(value = Errors.class, names = {"NONE", "COORDINATOR_NOT_AVAILABLE"})
      public void testClearFatalErrorWhenReceivingSuccessfulResponse(Errors error) {
          CoordinatorRequestManager coordinatorManager = setupCoordinatorManager(GROUP_ID);
          expectFindCoordinatorRequest(coordinatorManager, Errors.GROUP_AUTHORIZATION_FAILED);
          assertTrue(coordinatorManager.fatalError().isPresent());
  
          time.sleep(RETRY_BACKOFF_MS);
          // there are no successful responses, so the fatal error should persist
          assertTrue(coordinatorManager.fatalError().isPresent());
  
          // receiving a successful response should clear the fatal error
          expectFindCoordinatorRequest(coordinatorManager, error);
          assertTrue(coordinatorManager.fatalError().isEmpty());
      }
  
      private void expectFindCoordinatorRequest(
          CoordinatorRequestManager  coordinatorManager,
          Errors error
      ) {
          NetworkClientDelegate.PollResult res = coordinatorManager.poll(time.milliseconds());
          assertEquals(1, res.unsentRequests.size());
  
          NetworkClientDelegate.UnsentRequest unsentRequest = res.unsentRequests.get(0);
          unsentRequest.handler().onComplete(buildResponse(unsentRequest, error));
  
          boolean expectCoordinatorFound = error == Errors.NONE;
          assertEquals(expectCoordinatorFound, coordinatorManager.coordinator().isPresent());
      }
  
      private CoordinatorRequestManager setupCoordinatorManager(String groupId) {
          return new CoordinatorRequestManager(
              new LogContext(),
              RETRY_BACKOFF_MS,
              RETRY_BACKOFF_MS,
              groupId
          );
      }
  
      private ClientResponse buildResponse(
          NetworkClientDelegate.UnsentRequest request,
          Errors error
      ) {
          AbstractRequest abstractRequest = request.requestBuilder().build();
          assertInstanceOf(FindCoordinatorRequest.class, abstractRequest);
          FindCoordinatorRequest findCoordinatorRequest = (FindCoordinatorRequest) abstractRequest;
  
          FindCoordinatorResponse findCoordinatorResponse =
              FindCoordinatorResponse.prepareResponse(error, GROUP_ID, node);
          return new ClientResponse(
              new RequestHeader(ApiKeys.FIND_COORDINATOR, findCoordinatorRequest.version(), "", 1),
              request.handler(),
              node.idString(),
              time.milliseconds(),
              time.milliseconds(),
              false,
              null,
              null,
              findCoordinatorResponse
          );
      }
  }
  