Merge branch 'trunk' into KAFKA-16907

Author: frankvicky

.github/actions/run-gradle/action.yml

@@ -58,7 +58,7 @@ runs:
         timeout ${TIMEOUT_MINUTES}m ./gradlew --build-cache --continue --no-scan \
         -PtestLoggingEvents=started,passed,skipped,failed \
         -PmaxParallelForks=2 \
-        -PmaxTestRetries=1 -PmaxTestRetryFailures=3 \
+        -PmaxTestRetries=1 -PmaxTestRetryFailures=10 \
         -PmaxQuarantineTestRetries=3 -PmaxQuarantineTestRetryFailures=0 \
         -Pkafka.test.catalog.file=$TEST_CATALOG \
         -PcommitId=xxxxxxxxxxxxxxxx \
@@ -72,4 +72,4 @@ runs:
         name: ${{ inputs.build-scan-artifact-name }}
         path: ~/.gradle/build-scan-data
         compression-level: 9
-        if-no-files-found: ignore
+        if-no-files-found: ignore

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -2656,6 +2656,7 @@ CreateAclsRequest.Builder createRequest(int timeoutMs) {
 
             @Override
             void handleResponse(AbstractResponse abstractResponse) {
+                handleNotControllerError(abstractResponse);
                 CreateAclsResponse response = (CreateAclsResponse) abstractResponse;
                 List<AclCreationResult> responses = response.results();
                 Iterator<AclCreationResult> iter = responses.iterator();
@@ -2708,6 +2709,7 @@ DeleteAclsRequest.Builder createRequest(int timeoutMs) {
 
             @Override
             void handleResponse(AbstractResponse abstractResponse) {
+                handleNotControllerError(abstractResponse);
                 DeleteAclsResponse response = (DeleteAclsResponse) abstractResponse;
                 List<DeleteAclsResponseData.DeleteAclsFilterResult> results = response.filterResults();
                 Iterator<DeleteAclsResponseData.DeleteAclsFilterResult> iter = results.iterator();
@@ -2926,6 +2928,7 @@ public IncrementalAlterConfigsRequest.Builder createRequest(int timeoutMs) {
 
             @Override
             public void handleResponse(AbstractResponse abstractResponse) {
+                handleNotControllerError(abstractResponse);
                 IncrementalAlterConfigsResponse response = (IncrementalAlterConfigsResponse) abstractResponse;
                 Map<ConfigResource, ApiError> errors = IncrementalAlterConfigsResponse.fromResponseData(response.data());
                 for (Map.Entry<ConfigResource, KafkaFutureImpl<Void>> entry : futures.entrySet()) {
@@ -4089,8 +4092,11 @@ void handleFailure(Throwable throwable) {
     }
 
     private void handleNotControllerError(AbstractResponse response) throws ApiException {
+        // When sending requests directly to the follower controller, it might return NOT_LEADER_OR_FOLLOWER error.
         if (response.errorCounts().containsKey(Errors.NOT_CONTROLLER)) {
             handleNotControllerError(Errors.NOT_CONTROLLER);
+        } else if (metadataManager.usingBootstrapControllers() && response.errorCounts().containsKey(Errors.NOT_LEADER_OR_FOLLOWER)) {
+            handleNotControllerError(Errors.NOT_LEADER_OR_FOLLOWER);
         }
     }
 
@@ -4652,6 +4658,7 @@ DescribeQuorumRequest.Builder createRequest(int timeoutMs) {
 
             @Override
             void handleResponse(AbstractResponse response) {
+                handleNotControllerError(response);
                 final DescribeQuorumResponse quorumResponse = (DescribeQuorumResponse) response;
                 if (quorumResponse.data().errorCode() != Errors.NONE.code()) {
                     throw Errors.forCode(quorumResponse.data().errorCode()).exception(quorumResponse.data().errorMessage());
@@ -4849,6 +4856,7 @@ AddRaftVoterRequest.Builder createRequest(int timeoutMs) {
 
             @Override
             void handleResponse(AbstractResponse response) {
+                handleNotControllerError(response);
                 AddRaftVoterResponse addResponse = (AddRaftVoterResponse) response;
                 if (addResponse.data().errorCode() != Errors.NONE.code()) {
                     ApiError error = new ApiError(
@@ -4893,6 +4901,7 @@ RemoveRaftVoterRequest.Builder createRequest(int timeoutMs) {
 
             @Override
             void handleResponse(AbstractResponse response) {
+                handleNotControllerError(response);
                 RemoveRaftVoterResponse addResponse = (RemoveRaftVoterResponse) response;
                 if (addResponse.data().errorCode() != Errors.NONE.code()) {
                     ApiError error = new ApiError(

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java

@@ -159,7 +159,7 @@ public abstract class AbstractMembershipManager<R extends AbstractResponse> impl
     /**
      * If the member is currently leaving the group after a call to {@link #leaveGroup()} or
      * {@link #leaveGroupOnClose()}, this will have a future that will complete when the ongoing leave operation
-     * completes (callbacks executed and heartbeat request to leave is sent out). This will be empty is the
+     * completes (callbacks executed and heartbeat request to leave is sent out). This will be empty if the
      * member is not leaving.
      */
     private Optional<CompletableFuture<Void>> leaveGroupInProgress = Optional.empty();

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerMetadata.java

@@ -94,6 +94,6 @@ protected synchronized boolean retainTopic(String topic, boolean isInternal, lon
         if (isInternal && !includeInternalTopics)
             return false;
 
-        return subscription.matchesSubscribedPattern(topic);
+        return subscription.matchesSubscribedPattern(topic) || subscription.isAssignedFromRe2j(topic);
     }
 }