Merge remote-tracking branch 'apache-github/trunk' into kip-896-remove-old-protocol-versions

* apache-github/trunk: (25 commits)
  KAFKA-18270: FindCoordinator v0 incorrectly tagged as deprecated (apache#18262)
  KAFKA-18284: Add group coordinator records for Streams rebalance protocol (apache#18228)
  MINOR: Fix flaky state updater test (apache#18253)
  MINOR: improve StreamsResetter logging (apache#18237)
  KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade (apache#18176)
  Add IT for share consumer with duration base offet auto reset (apache#18251)
  KAFKA-18283: Add StreamsGroupDescribe RPC definitions (apache#18230)
  KAFKA-18241: add docs check to CI (apache#18183)
  KAFKA-18223 Improve flaky test report (apache#18212)
  MINOR Remove triage label in nightly job (apache#18147)
  KAFKA-18294 Remove deprecated SourceTask#commitRecord (apache#18260)
  KAFKA-18264 Remove NotLeaderForPartitionException (apache#18211)
  KAFKA-13722: Refactor SerdeGetter (apache#18242)
  KAFKA-18094 Remove deprecated TopicListing(String, Boolean) (apache#18248)
  KAFKA-18282: Add StreamsGroupHeartbeat RPC definitions (apache#18227)
  KAFKA-18026: KIP-1112 migrate KTableSuppressProcessorSupplier (apache#18150)
  KAFKA-18026: transition KTable#filter impl to use processor wrapper (apache#18205)
  KAFKA-18293 Remove `org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler` and `org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerValidatorCallbackHandler` (apache#18244)
  MINOR: add assertion about groupEpoch and targetAssignmentEpoch to testConsumerGroups (apache#18203)
  KAFKA-17960; PlaintextAdminIntegrationTest.testConsumerGroups fails with CONSUMER group protocol (apache#18234)
  ...

Author: ijuma

.github/scripts/develocity_reports.py

@@ -67,13 +67,11 @@ Unlike trunk, the PR builds _will_ utilize the Gradle cache.
 ### PR Triage
 
 In order to get the attention of committers, we have a triage workflow for Pull Requests
-opened by non-committers. This workflow consists of three files:
+opened by non-committers. This workflow consists of two files:
 
-* [pr-update.yml](pr-update.yml) When a PR is created add the `triage` label if the PR
+* [pr-update.yml](pr-update.yml) When a PR is created, add the `triage` label if the PR
   was opened by a non-committer.
-* [pr-reviewed-trigger.yml](pr-reviewed-trigger.yml) Runs when any PR is reviewed. 
-  Used as a trigger for the next workflow
-* [pr-reviewed.yml](pr-reviewed.yml) Remove the `triage` label after a PR has been reviewed
+* [pr-reviewed.yml](pr-reviewed.yml) Cron job to remove the `triage` label from PRs which have been reviewed
 
 _The pr-update.yml workflow includes pull_request_target!_
 
@@ -100,7 +98,7 @@ There are two files related to this workflow:
 
 * [pr-labeled.yml](pr-labeled.yml) approves a pending approval for PRs that have
 been labeled with `ci-approved`
-* [ci-requested.yml](ci-requested.yml) approves future CI requests automatically
+* [ci-requested.yml](ci-requested.yml) approves future workflow requests automatically
 if the PR has the `ci-approved` label
 
 _The pr-labeled.yml workflow includes pull_request_target!_

.github/workflows/README.md

@@ -117,7 +117,7 @@ jobs:
         # --scan:         Publish the build scan. This will only work on PRs from apache/kafka and trunk
         # --no-scan:      For public fork PRs, we won't attempt to publish the scan
         run: |
-          ./gradlew --build-cache --info $SCAN_ARG check -x test
+          ./gradlew --build-cache --info $SCAN_ARG check siteDocTar -x test
       - name: Archive check reports
         if: always()
         uses: actions/upload-artifact@v4
@@ -137,6 +137,16 @@ jobs:
         run: python .github/scripts/rat.py
         env:
           GITHUB_WORKSPACE: ${{ github.workspace }}
+      - name: Check generated documentation
+        # Check if there are any empty files under ./site-docs/generated, If any empty files are found, print an error 
+        # message and list the empty files
+        run: |
+          tar zxvf core/build/distributions/kafka_2.13-$(./gradlew properties | grep version: | awk '{print $NF}' | head -n 1)-site-docs.tgz
+          if find ./site-docs/generated -type f -exec grep -L "." {} \; | grep -q "."; then
+            echo "One or more documentation files are empty!" >&2
+            find ./site-docs/generated -type f -exec grep -L "." {} \; >&2
+            exit 1
+          fi
 
   test:
     needs: [validate, load-catalog]

.github/workflows/build.yml

@@ -20,7 +20,6 @@ on:
   schedule:
     - cron: '0 6 * * *'   # Run daily at 6am UTC
 
-
 jobs:
   flaky-test-report:
     name: Flaky Test Report
@@ -38,6 +37,6 @@ jobs:
         uses: ./.github/actions/setup-python
       - name: Run Report
         env:
-          GE_ACCESS_TOKEN: ${{ secrets.GE_ACCESS_TOKEN }}
+          DEVELOCITY_ACCESS_TOKEN: ${{ secrets.DV_API_ACCESS }}
         run: |
-          python ./.github/scripts/develocity_reports.py >> $GITHUB_STEP_SUMMARY
+          python ./.github/scripts/develocity_reports.py >> $GITHUB_STEP_SUMMARY

.github/workflows/generate-reports.yml

@@ -16,39 +16,48 @@
 name: Remove Triage Label
 
 on:
-  workflow_run:
-    workflows: [Pull Request Reviewed]
-    types:
-      - completed
+  workflow_dispatch:      # Let us run manually
+
+  schedule:
+    - cron: '0 3 * * *'    # Run at 3:00 UTC nightly -- just before the "stale.yml" workflow
 
 jobs:
-  # This job runs with elevated permissions and the ability to modify pull requests. The steps taken here
-  # should be limited to updating labels and adding comments to PRs. This approach is taken from
-  # https://securitylab.github.com/resources/github-actions-preventing-pwn-requests/.
   remove-triage:
-    if: ${{ github.event.workflow_run.conclusion == 'success' }}
     runs-on: ubuntu-latest
     steps:
       - name: Env
         run: printenv
         env:
           GITHUB_CONTEXT: ${{ toJson(github) }}
-      - uses: actions/download-artifact@v4
-        with:
-          github-token: ${{ github.token }}
-          run-id: ${{ github.event.workflow_run.id }}
-          name: pr-number.txt
       - name: Remove label
         uses: actions/github-script@v7
         continue-on-error: true
         with:
           github-token: ${{ secrets.GITHUB_TOKEN }}
           script: |
-            var fs = require('fs');
-            var pr_number = Number(fs.readFileSync('./pr-number.txt'));
-            await github.rest.issues.removeLabel({
-              owner: context.repo.owner,
-              repo: context.repo.repo,
-              issue_number: pr_number,
-              name: 'triage'
+            github.paginate("GET /search/issues{?q}", {
+              q: "repo:apache/kafka label:triage is:pull-request"
+            })
+            .then((pulls) => {
+              pulls.forEach(pull => {
+                github.request("GET /repos/{owner}/{repo}/pulls/{pull_number}/reviews", {
+                  owner: "apache",
+                  repo: "kafka",
+                  pull_number: pull.number,
+                  headers: {
+                    "X-GitHub-Api-Version": "2022-11-28"
+                  }
+                }).then((resp) => {
+                  console.log("Found " + resp.data.length + " reviews for PR " + pull.number);
+                  if (resp.data.length > 0) {
+                    console.log("Removing 'triage' label from PR " + pull.number + " : " + pull.title);
+                    github.rest.issues.removeLabel({
+                      owner: "apache",
+                      repo: "kafka",
+                      issue_number: pull.number,
+                      name: "triage"
+                    });
+                  }
+                });
+              });
             });

.github/workflows/pr-reviewed-trigger.yml

@@ -27,20 +27,6 @@ public class TopicListing {
     private final Uuid topicId;
     private final boolean internal;
 
-    /**
-     * Create an instance with the specified parameters.
-     *
-     * @param name The topic name
-     * @param internal Whether the topic is internal to Kafka
-     * @deprecated Since 3.0 use {@link #TopicListing(String, Uuid, boolean)} instead
-     */
-    @Deprecated
-    public TopicListing(String name, boolean internal) {
-        this.name = name;
-        this.internal = internal;
-        this.topicId = Uuid.ZERO_UUID;
-    }
-
     /**
      * Create an instance with the specified parameters.
      *

.github/workflows/pr-reviewed.yml

@@ -665,6 +665,7 @@ public void initTransactions() {
         sender.wakeup();
         result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS);
         producerMetrics.recordInit(time.nanoseconds() - now);
+        transactionManager.maybeUpdateTransactionV2Enabled(true);
     }
 
     /**

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

@@ -211,7 +211,7 @@ private boolean isTransitionValid(State source, State target) {
                 case UNINITIALIZED:
                     return source == READY || source == ABORTABLE_ERROR;
                 case INITIALIZING:
-                    return source == UNINITIALIZED || source == ABORTING_TRANSACTION;
+                    return source == UNINITIALIZED || source == COMMITTING_TRANSACTION || source == ABORTING_TRANSACTION;
                 case READY:
                     return source == INITIALIZING || source == COMMITTING_TRANSACTION || source == ABORTING_TRANSACTION;
                 case IN_TRANSACTION:
@@ -312,7 +312,6 @@ public synchronized void beginTransaction() {
         throwIfPendingState("beginTransaction");
         maybeFailWithError();
         transitionTo(State.IN_TRANSACTION);
-        maybeUpdateTransactionV2Enabled();
     }
 
     public synchronized TransactionalRequestResult beginCommit() {
@@ -348,10 +347,17 @@ private TransactionalRequestResult beginCompletingTransaction(TransactionResult
             isTransactionV2Enabled
         );
 
+        // Maybe update the transaction version here before we enqueue the EndTxn request so there are no races with
+        // completion of the EndTxn request. Since this method may update clientSideEpochBumpRequired, we want to update
+        // before the check below, but we also want to call it after the EndTxnRequest.Builder so we complete the transaction
+        // with the same version as it started.
+        maybeUpdateTransactionV2Enabled(false);
+
         EndTxnHandler handler = new EndTxnHandler(builder);
         enqueueRequest(handler);
 
         // If an epoch bump is required for recovery, initialize the transaction after completing the EndTxn request.
+        // If we are upgrading to TV2 transactions on the next transaction, also bump the epoch.
         if (clientSideEpochBumpRequired) {
             return initializeTransactions(this.producerIdAndEpoch);
         }
@@ -437,15 +443,25 @@ public boolean isTransactional() {
         return transactionalId != null;
     }
 
-    // Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
-    public synchronized void maybeUpdateTransactionV2Enabled() {
+    /**
+     *  Check all the finalized features from apiVersions to verify whether the transaction V2 is enabled.
+     *  Sets clientSideEpochBumpRequired if upgrading to V2 since we need to bump the epoch.
+     *  This is because V2 no longer adds partitions explicitly and there are some edge cases on upgrade
+     *  that can be avoided by fencing the old V1 transaction epoch. For example, we won't consider
+     *  partitions from the previous transaction as already added to the new V2 transaction if the epoch is fenced.
+     */
+
+    public synchronized void maybeUpdateTransactionV2Enabled(boolean onInitiatialization) {
         if (latestFinalizedFeaturesEpoch >= apiVersions.getMaxFinalizedFeaturesEpoch()) {
             return;
         }
         ApiVersions.FinalizedFeaturesInfo info = apiVersions.getFinalizedFeaturesInfo();
         latestFinalizedFeaturesEpoch = info.finalizedFeaturesEpoch;
         Short transactionVersion = info.finalizedFeatures.get("transaction.version");
+        boolean wasTransactionV2Enabled = isTransactionV2Enabled;
         isTransactionV2Enabled = transactionVersion != null && transactionVersion >= 2;
+        if (!onInitiatialization && !wasTransactionV2Enabled && isTransactionV2Enabled)
+            clientSideEpochBumpRequired = true;
     }
 
     public boolean isTransactionV2Enabled() {
@@ -1652,6 +1668,8 @@ public void handleResponse(AbstractResponse response) {
                 // When Transaction Version 2 is enabled, the end txn request 5+ is used,
                 // it mandates bumping the epoch after every transaction.
                 // If the epoch overflows, a new producerId is returned with epoch set to 0.
+                // Note, we still may see EndTxn TV1 (< 5) responses when the producer has upgraded to TV2 due to the upgrade
+                // occurring at the end of beginCompletingTransaction. The next transaction started should be TV2.
                 if (endTxnResponse.data().producerId() != -1) {
                     ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(
                         endTxnResponse.data().producerId(),