Merge branch 'apache:develop' into develop

.github/workflows/bazel.yml

@@ -8,9 +8,6 @@ on:
  - develop
  - bazel
 
-permissions:
- actions: write
-
 jobs:
  build:
  name: "bazel-compile (${{ matrix.os }})"
@@ -23,14 +20,4 @@ jobs:
  - name: Build
  run: bazel build --config=remote //...
  - name: Run Tests
- run: bazel test --config=remote //...
- - name: Retry if failed
- # if it failed , retry 2 times at most
- if: failure() && fromJSON(github.run_attempt) < 3
- continue-on-error: true
- env:
- GH_REPO: ${{ github.repository }}
- GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
- run: |
- echo "Attempting to retry workflow..."
- gh workflow run rerun-workflow.yml -F run_id=${{ github.run_id }}
+ run: bazel test --config=remote //...

.github/workflows/maven.yaml

@@ -5,9 +5,6 @@ on:
  push:
  branches: [master, develop, bazel]
 
-permissions:
- actions: write
-
 jobs:
  java_build:
  name: "maven-compile (${{ matrix.os }}, JDK-${{ matrix.jdk }})"
@@ -44,15 +41,4 @@ jobs:
  with:
  name: jvm-crash-logs
  path: /Users/runner/work/rocketmq/rocketmq/broker/hs_err_pid*.log
- retention-days: 1
-
- - name: Retry if failed
- # if it failed , retry 2 times at most
- if: failure() && fromJSON(github.run_attempt) < 3
- continue-on-error: true
- env:
- GH_REPO: ${{ github.repository }}
- GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
- run: |
- echo "Attempting to retry workflow..."
- gh workflow run rerun-workflow.yml -F run_id=${{ github.run_id }}
+ retention-days: 1

.github/workflows/pr-ci.yml

@@ -21,7 +21,7 @@ jobs:
  - name: Build distribution tar
  run: |
  mvn -Prelease-all -DskipTests -Dspotbugs.skip=true clean install -U
- - uses: actions/upload-artifact@v3
+ - uses: actions/upload-artifact@v4
  name: Upload distribution tar
  with:
  name: rocketmq
@@ -30,7 +30,7 @@ jobs:
  run: |
  mkdir -p ./pr
  echo ${{ github.event.number }} > ./pr/NR
- - uses: actions/upload-artifact@v2
+ - uses: actions/upload-artifact@v4
  with:
  name: pr
  path: pr/

.github/workflows/pr-e2e-test.yml

@@ -25,18 +25,18 @@ jobs:
  java-version: ["8"]
  steps:
  - name: 'Download artifact'
- uses: actions/github-script@v3.1.0
+ uses: actions/github-script@v6
  with:
  script: |
- var artifacts = await github.actions.listWorkflowRunArtifacts({
+ let artifacts = await github.rest.actions.listWorkflowRunArtifacts({
  owner: context.repo.owner,
  repo: context.repo.repo,
  run_id: ${{github.event.workflow_run.id }},
  });
- var matchArtifactRmq = artifacts.data.artifacts.filter((artifact) => {
+ let matchArtifactRmq = artifacts.data.artifacts.filter((artifact) => {
  return artifact.name == "rocketmq"
  })[0];
- var download = await github.actions.downloadArtifact({
+ let download = await github.rest.actions.downloadArtifact({
  owner: context.repo.owner,
  repo: context.repo.repo,
  artifact_id: matchArtifactRmq.id,
@@ -68,7 +68,7 @@ jobs:
  mkdir versionlist
  touch versionlist/"${version}-`echo ${{ matrix.base-image }} | sed -e "s/:/-/g"`"
  sh ./build-image-local.sh ${version} ${{ matrix.base-image }} ${{ matrix.java-version }} ${DOCKER_REPO}
- - uses: actions/upload-artifact@v3
+ - uses: actions/upload-artifact@v4
  name: Upload distribution tar
  with:
  name: versionlist
@@ -85,7 +85,7 @@ jobs:
  outputs:
  version-json: ${{ steps.show_versions.outputs.version-json }}
  steps:
- - uses: actions/download-artifact@v3
+ - uses: actions/download-artifact@v4
  name: Download versionlist
  with:
  name: versionlist
@@ -96,6 +96,7 @@ jobs:
  a=(`ls versionlist`)
  printf '%s\n' "${a[@]}" | jq -R . | jq -s .
  echo version-json=`printf '%s\n' "${a[@]}" | jq -R . | jq -s .` >> $GITHUB_OUTPUT
+
  deploy:
  if: ${{ success() }}
  name: Deploy RocketMQ
@@ -158,7 +159,7 @@ jobs:
  annotate_only: true
  include_passed: true
  detailed_summary: true
- - uses: actions/upload-artifact@v3
+ - uses: actions/upload-artifact@v4
  if: always()
  name: Upload test log
  with:
@@ -199,7 +200,7 @@ jobs:
  annotate_only: true
  include_passed: true
  detailed_summary: true
- - uses: actions/upload-artifact@v3
+ - uses: actions/upload-artifact@v4
  if: always()
  name: Upload test log
  with:
@@ -235,7 +236,7 @@ jobs:
  annotate_only: true
  include_passed: true
  detailed_summary: true
- - uses: actions/upload-artifact@v3
+ - uses: actions/upload-artifact@v4
  if: always()
  name: Upload test log
  with:
@@ -258,5 +259,4 @@ jobs:
  action: "clean"
  ask-config: "${{ secrets.ASK_CONFIG_VIRGINA }}"
  test-version: "${{ matrix.version }}"
- job-id: ${{ strategy.job-index }}
-
+ job-id: ${{ strategy.job-index }}

.github/workflows/push-ci.yml

@@ -31,7 +31,7 @@ jobs:
  - name: Build distribution tar
  run: |
  mvn -Prelease-all -DskipTests -Dspotbugs.skip=true clean install -U
- - uses: actions/upload-artifact@v3
+ - uses: actions/upload-artifact@v4
  name: Upload distribution tar
  with:
  name: rocketmq
@@ -53,7 +53,7 @@ jobs:
  repository: apache/rocketmq-docker.git
  ref: master
  path: rocketmq-docker
- - uses: actions/download-artifact@v3
+ - uses: actions/download-artifact@v4
  name: Download distribution tar
  with:
  name: rocketmq
@@ -72,7 +72,7 @@ jobs:
  mkdir versionlist
  touch versionlist/"${version}-`echo ${{ matrix.base-image }} | sed -e "s/:/-/g"`"
  sh ./build-image-local.sh ${version} ${{ matrix.base-image }} ${{ matrix.java-version }} ${DOCKER_REPO}
- - uses: actions/upload-artifact@v3
+ - uses: actions/upload-artifact@v4
  name: Upload distribution tar
  with:
  name: versionlist
@@ -90,7 +90,7 @@ jobs:
  outputs:
  version-json: ${{ steps.show_versions.outputs.version-json }}
  steps:
- - uses: actions/download-artifact@v3
+ - uses: actions/download-artifact@v4
  name: Download versionlist
  with:
  name: versionlist
@@ -163,7 +163,7 @@ jobs:
  annotate_only: true
  include_passed: true
  detailed_summary: true
- - uses: actions/upload-artifact@v3
+ - uses: actions/upload-artifact@v4
  if: always()
  name: Upload test log
  with:
@@ -204,7 +204,7 @@ jobs:
  annotate_only: true
  include_passed: true
  detailed_summary: true
- - uses: actions/upload-artifact@v3
+ - uses: actions/upload-artifact@v4
  if: always()
  name: Upload test log
  with:
@@ -240,7 +240,7 @@ jobs:
  annotate_only: true
  include_passed: true
  detailed_summary: true
- - uses: actions/upload-artifact@v3
+ - uses: actions/upload-artifact@v4
  if: always()
  name: Upload test log
  with:

.github/workflows/rerun-workflow.yml

@@ -1,21 +1,22 @@
 name: Rerun workflow
 on:
- workflow_dispatch:
- inputs:
-  run_id:
-  required: true
+ workflow_run:
+ workflows: ["Build and Run Tests by Maven" , "Build and Run Tests by Bazel"]
+ types:
+ - completed
 
 permissions:
  actions: write
 
 jobs:
  rerun:
+ if: github.event.workflow_run.conclusion == 'failure' && fromJSON(github.event.workflow_run.run_attempt) < 3
  runs-on: ubuntu-latest
  steps:
- - name: rerun ${{ inputs.run_id }}
+ - name: rerun ${{  github.event.workflow_run.id }}
  env:
- GH_REPO: ${{ github.repository }}
+ GH_REPO:  ${{ github.repository }}
  GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
  run: |
- gh run watch ${{ inputs.run_id }} > /dev/null 2>&1
- gh run rerun ${{ inputs.run_id }} --failed
+ gh run watch ${{ github.event.workflow_run.id }} > /dev/null 2>&1
+ gh run rerun ${{ github.event.workflow_run.id }} --failed

.github/workflows/snapshot-automation.yml

@@ -69,7 +69,7 @@ jobs:
  MAVEN_SETTINGS: ${{ github.workspace }}/.github/asf-deploy-settings.xml
  run: |
  mvn -Prelease-all -DskipTests -Dspotbugs.skip=true clean install -U
- - uses: actions/upload-artifact@v3
+ - uses: actions/upload-artifact@v4
  name: Upload distribution tar
  with:
  name: rocketmq
@@ -91,7 +91,7 @@ jobs:
  repository: apache/rocketmq-docker.git
  ref: master
  path: rocketmq-docker
- - uses: actions/download-artifact@v3
+ - uses: actions/download-artifact@v4
  name: Download distribution tar
  with:
  name: rocketmq
@@ -110,7 +110,7 @@ jobs:
  mkdir versionlist
  touch versionlist/"${version}-`echo ${{ matrix.base-image }} | sed -e "s/:/-/g"`"
  sh ./build-image-local.sh ${version} ${{ matrix.base-image }} ${{ matrix.java-version }} ${DOCKER_REPO}
- - uses: actions/upload-artifact@v3
+ - uses: actions/upload-artifact@v4
  name: Upload distribution tar
  with:
  name: versionlist
@@ -125,7 +125,7 @@ jobs:
  outputs:
  version-json: ${{ steps.show_versions.outputs.version-json }}
  steps:
- - uses: actions/download-artifact@v3
+ - uses: actions/download-artifact@v4
  name: Download versionlist
  with:
  name: versionlist
@@ -200,7 +200,7 @@ jobs:
  annotate_only: true
  include_passed: true
  detailed_summary: true
- - uses: actions/upload-artifact@v3
+ - uses: actions/upload-artifact@v4
  if: always()
  name: Upload test log
  with:

acl/pom.xml

@@ -13,7 +13,7 @@
  <parent>
  <groupId>org.apache.rocketmq</groupId>
  <artifactId>rocketmq-all</artifactId>
- <version>5.3.1-SNAPSHOT</version>
+ <version>5.3.2-SNAPSHOT</version>
  </parent>
  <artifactId>rocketmq-acl</artifactId>
  <name>rocketmq-acl ${project.version}</name>

auth/pom.xml

@@ -13,7 +13,7 @@
  <parent>
  <groupId>org.apache.rocketmq</groupId>
  <artifactId>rocketmq-all</artifactId>
- <version>5.3.1-SNAPSHOT</version>
+ <version>5.3.2-SNAPSHOT</version>
  </parent>
  <artifactId>rocketmq-auth</artifactId>
  <name>rocketmq-auth ${project.version}</name>

auth/src/main/java/org/apache/rocketmq/auth/authorization/factory/AuthorizationFactory.java

@@ -105,7 +105,7 @@ public static AuthorizationEvaluator getEvaluator(AuthConfig config, Supplier<?>
  public static AuthorizationStrategy getStrategy(AuthConfig config, Supplier<?> metadataService) {
  try {
  Class<? extends AuthorizationStrategy> clazz = StatelessAuthorizationStrategy.class;
- if (StringUtils.isNotBlank(config.getAuthenticationStrategy())) {
+ if (StringUtils.isNotBlank(config.getAuthorizationStrategy())) {
  clazz = (Class<? extends AuthorizationStrategy>) Class.forName(config.getAuthorizationStrategy());
  }
  return clazz.getDeclaredConstructor(AuthConfig.class, Supplier.class).newInstance(config, metadataService);

broker/pom.xml

@@ -13,7 +13,7 @@
  <parent>
  <groupId>org.apache.rocketmq</groupId>
  <artifactId>rocketmq-all</artifactId>
- <version>5.3.1-SNAPSHOT</version>
+ <version>5.3.2-SNAPSHOT</version>
  </parent>
 
  <modelVersion>4.0.0</modelVersion>

broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java

@@ -18,7 +18,6 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Lists;
-import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.util.AbstractMap;
 import java.util.ArrayList;
@@ -789,6 +788,9 @@ public boolean initializeMessageStore() {
  defaultMessageStore = new RocksDBMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener, this.brokerConfig, topicConfigManager.getTopicConfigTable());
  } else {
  defaultMessageStore = new DefaultMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener, this.brokerConfig, topicConfigManager.getTopicConfigTable());
+ if (messageStoreConfig.isRocksdbCQDoubleWriteEnable()) {
+ defaultMessageStore.enableRocksdbCQWrite();
+ }
  }
 
  if (messageStoreConfig.isEnableDLegerCommitLog()) {
@@ -812,7 +814,7 @@ public boolean initializeMessageStore() {
  this.timerMessageStore.registerEscapeBridgeHook(msg -> escapeBridge.putMessage(msg));
  this.messageStore.setTimerMessageStore(this.timerMessageStore);
  }
- } catch (IOException e) {
+ } catch (Exception e) {
  result = false;
  LOG.error("BrokerController#initialize: unexpected error occurs", e);
  }

broker/src/main/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManager.java

@@ -31,6 +31,7 @@
 import org.apache.rocketmq.broker.BrokerController;
 import org.apache.rocketmq.broker.BrokerPathConfigHelper;
 import org.apache.rocketmq.common.ConfigManager;
+import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.logging.org.slf4j.Logger;
@@ -373,6 +374,25 @@ public void setDataVersion(DataVersion dataVersion) {
  this.dataVersion = dataVersion;
  }
 
+ public boolean loadDataVersion() {
+ String fileName = null;
+ try {
+ fileName = this.configFilePath();
+ String jsonString = MixAll.file2String(fileName);
+ if (jsonString != null) {
+ ConsumerOffsetManager obj = RemotingSerializable.fromJson(jsonString, ConsumerOffsetManager.class);
+ if (obj != null) {
+ this.dataVersion = obj.dataVersion;
+ }
+ LOG.info("load consumer offset dataVersion success,{},{} ", fileName, jsonString);
+ }
+ return true;
+ } catch (Exception e) {
+ LOG.error("load consumer offset dataVersion failed " + fileName, e);
+ return false;
+ }
+ }
+
  public void removeOffset(final String group) {
  Iterator<Entry<String, ConcurrentMap<Integer, Long>>> it = this.offsetTable.entrySet().iterator();
  while (it.hasNext()) {