diff --git a/.github/workflows/coverage.yml b/.github/workflows/coverage.yml index afa8e0f51a..b249072f8d 100644 --- a/.github/workflows/coverage.yml +++ b/.github/workflows/coverage.yml @@ -10,10 +10,10 @@ jobs: steps: - uses: actions/checkout@master - name: Set up JDK 8 - uses: actions/setup-java@v2 + uses: actions/setup-java@v4 with: java-version: "8" - distribution: "adopt" + distribution: "corretto" cache: "maven" - name: Generate coverage report run: mvn -B test -T 2C --file pom.xml diff --git a/.github/workflows/integration-test.yml b/.github/workflows/integration-test.yml index 8179f36287..ad5bcbd6c2 100644 --- a/.github/workflows/integration-test.yml +++ b/.github/workflows/integration-test.yml @@ -26,10 +26,10 @@ jobs: uses: actions/checkout@v2 - name: Set up JDK ${{ matrix.jdk }} - uses: actions/setup-java@v2 + uses: actions/setup-java@v4 with: java-version: ${{ matrix.jdk }} - distribution: "adopt" + distribution: "corretto" cache: "maven" - name: Run integration tests with Maven diff --git a/.github/workflows/maven.yaml b/.github/workflows/maven.yaml index d0c0ba7d9f..4eacd65b84 100644 --- a/.github/workflows/maven.yaml +++ b/.github/workflows/maven.yaml @@ -19,10 +19,12 @@ jobs: - name: Checkout uses: actions/checkout@v2 - name: Set up JDK ${{ matrix.jdk }} - uses: actions/setup-java@v2 + uses: actions/setup-java@v4 with: java-version: ${{ matrix.jdk }} - distribution: "adopt" + # See https://github.com/actions/setup-java?tab=readme-ov-file#supported-distributions + # AdoptOpenJDK got moved to Eclipse Temurin and won't be updated anymore. + distribution: "corretto" cache: "maven" - name: Build with Maven run: mvn -B package --file pom.xml @@ -41,4 +43,4 @@ jobs: with: name: jvm-crash-logs path: /Users/runner/work/rocketmq/rocketmq/broker/hs_err_pid*.log - retention-days: 1 \ No newline at end of file + retention-days: 1 diff --git a/.github/workflows/snapshot-automation.yml b/.github/workflows/snapshot-automation.yml index 9fb16cb13c..2c4ede4b6e 100644 --- a/.github/workflows/snapshot-automation.yml +++ b/.github/workflows/snapshot-automation.yml @@ -59,9 +59,9 @@ jobs: with: ref: ${{ github.event.inputs.branch }} - - uses: actions/setup-java@v3 + - uses: actions/setup-java@v4 with: - distribution: "temurin" + distribution: "corretto" java-version: "8" cache: "maven" - name: Build distribution tar @@ -238,10 +238,10 @@ jobs: ref: develop persist-credentials: false - name: Set up JDK - uses: actions/setup-java@v3 + uses: actions/setup-java@v4 with: java-version: 8 - distribution: "temurin" + distribution: "corretto" cache: "maven" - name: Update default pom version if: github.event.inputs.rocketmq_version == '' diff --git a/store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueOffsetTable.java b/store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueOffsetTable.java index 889131d1cc..a91ae5e244 100644 --- a/store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueOffsetTable.java +++ b/store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueOffsetTable.java @@ -85,7 +85,7 @@ public class RocksDBConsumeQueueOffsetTable { * │ (4 Bytes) │ (1 Bytes) │ (1 Bytes) │ (3 Bytes) │ (1 Bytes) │ (4 Bytes) │ * ├─────────────────────────┴───────────┴───────────┴───────────┴───────────┴─────────────┤ */ - private static final int OFFSET_KEY_LENGTH_WITHOUT_TOPIC_BYTES = 4 + 1 + 1 + 3 + 1 + 4; + public static final int OFFSET_KEY_LENGTH_WITHOUT_TOPIC_BYTES = 4 + 1 + 1 + 3 + 1 + 4; private static final int OFFSET_VALUE_LENGTH = 8 + 8; /** @@ -682,7 +682,7 @@ public class RocksDBConsumeQueueOffsetTable { return byteBuffer; } - private static void buildOffsetKeyByteBuffer(final ByteBuffer byteBuffer, final byte[] topicBytes, + public static void buildOffsetKeyByteBuffer(final ByteBuffer byteBuffer, final byte[] topicBytes, final int queueId, final boolean max) { byteBuffer.position(0).limit(OFFSET_KEY_LENGTH_WITHOUT_TOPIC_BYTES + topicBytes.length); buildOffsetKeyByteBuffer0(byteBuffer, topicBytes, queueId, max); diff --git a/store/src/test/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueOffsetTableTest.java b/store/src/test/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueOffsetTableTest.java new file mode 100644 index 0000000000..b1e12d4946 --- /dev/null +++ b/store/src/test/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueOffsetTableTest.java @@ -0,0 +1,131 @@ +/* + * 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.rocketmq.store.queue; + +import java.io.File; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.rocketmq.store.DefaultMessageStore; +import org.apache.rocketmq.store.queue.offset.OffsetEntryType; +import org.apache.rocketmq.store.rocksdb.ConsumeQueueRocksDBStorage; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.Mockito; +import org.mockito.junit.MockitoJUnitRunner; +import org.rocksdb.Options; +import org.rocksdb.RocksDB; +import org.rocksdb.RocksDBException; +import org.rocksdb.RocksIterator; +import org.rocksdb.WriteBatch; +import org.rocksdb.WriteOptions; + +@RunWith(MockitoJUnitRunner.class) +public class RocksDBConsumeQueueOffsetTableTest { + + private RocksDBConsumeQueueOffsetTable offsetTable; + + @Mock + private ConsumeQueueRocksDBStorage rocksDBStorage; + + @Mock + private RocksDBConsumeQueueTable consumeQueueTable; + + @Mock + private DefaultMessageStore messageStore; + + private static RocksDB db; + + private static File dbPath; + + private static String topicName; + + @BeforeClass + public static void initDB() throws IOException, RocksDBException { + TemporaryFolder tempFolder = new TemporaryFolder(); + tempFolder.create(); + dbPath = tempFolder.newFolder(); + + db = RocksDB.open(dbPath.getAbsolutePath()); + StringBuilder topicBuilder = new StringBuilder(); + for (int i = 0; i < 100; i++) { + topicBuilder.append("topic"); + } + topicName = topicBuilder.toString(); + byte[] topicInBytes = topicName.getBytes(StandardCharsets.UTF_8); + + ByteBuffer keyBuffer = ByteBuffer.allocateDirect(RocksDBConsumeQueueOffsetTable.OFFSET_KEY_LENGTH_WITHOUT_TOPIC_BYTES + topicInBytes.length); + RocksDBConsumeQueueOffsetTable.buildOffsetKeyByteBuffer(keyBuffer, topicInBytes, 1, true); + Assert.assertEquals(0, keyBuffer.position()); + Assert.assertEquals(RocksDBConsumeQueueOffsetTable.OFFSET_KEY_LENGTH_WITHOUT_TOPIC_BYTES + topicInBytes.length, keyBuffer.limit()); + + ByteBuffer valueBuffer = ByteBuffer.allocateDirect(Long.BYTES + Long.BYTES); + valueBuffer.putLong(100); + valueBuffer.putLong(2); + valueBuffer.flip(); + + try (WriteBatch writeBatch = new WriteBatch(); + WriteOptions writeOptions = new WriteOptions()) { + writeOptions.setDisableWAL(false); + writeOptions.setSync(true); + writeBatch.put(keyBuffer, valueBuffer); + db.write(writeOptions, writeBatch); + } + + } + + @AfterClass + public static void tearDownDB() throws RocksDBException { + db.closeE(); + RocksDB.destroyDB(dbPath.getAbsolutePath(), new Options()); + } + + @Before + public void setUp() { + RocksIterator iterator = db.newIterator(); + Mockito.doReturn(iterator).when(rocksDBStorage).seekOffsetCF(); + offsetTable = new RocksDBConsumeQueueOffsetTable(consumeQueueTable, rocksDBStorage, messageStore); + } + + /** + * Verify forEach can expand key-buffer properly and works well for long topic names. + * + * @throws RocksDBException If there is an RocksDB error. + */ + @Test + public void testForEach() throws RocksDBException { + AtomicBoolean called = new AtomicBoolean(false); + offsetTable.forEach(entry -> true, entry -> { + called.set(true); + Assert.assertEquals(topicName, entry.topic); + Assert.assertTrue(topicName.length() > 256); + Assert.assertEquals(1, entry.queueId); + Assert.assertEquals(100, entry.commitLogOffset); + Assert.assertEquals(2, entry.offset); + Assert.assertEquals(OffsetEntryType.MAXIMUM, entry.type); + }); + Assert.assertTrue(called.get()); + } +}