From bd0e9c09db9748f7f74a0c707579142dccf30afc Mon Sep 17 00:00:00 2001
From: PiteXChen <44110731+RapperCL@users.noreply.github.com>
Date: Tue, 29 Aug 2023 19:39:27 +0800
Subject: [PATCH 1/7] [ISSUE #7111] Remove responseFuture from the
responseTable when exception occurs (#7112)
* remove responseFuture when exception
* Empty-Commit
Co-authored-by: chenyong152 <chenyong152@midea.com>
.../apache/rocketmq/remoting/netty/NettyRemotingAbstract.java | 1 +
1 file changed, 1 insertion(+)
@@ -529,6 +529,7 @@ public abstract class NettyRemotingAbstract {
log.warn("send a request command to channel <{}> failed.", RemotingHelper.parseChannelRemoteAddr(channel));
});
} catch (Exception e) {
+ responseTable.remove(opaque);
responseFuture.release();
log.warn("send a request command to channel <" + RemotingHelper.parseChannelRemoteAddr(channel) + "> Exception", e);
throw new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), e);
--
2.32.0.windows.2
From c78061bf6ca5f35452510ec4107c46735c51c316 Mon Sep 17 00:00:00 2001
From: lizhimins <707364882@qq.com>
Date: Wed, 30 Aug 2023 22:29:51 +0800
Subject: [PATCH 2/7] [ISSUE#7280] Fix and refactor handle commit exception in
tiered storage (#7281)
* refactor handle commit exception
* refactor handle commit exception
* fix handle commit exception
.../tieredstore/TieredDispatcher.java | 3 +-
.../tieredstore/TieredMessageFetcher.java | 57 ++--
.../tieredstore/TieredMessageStore.java | 26 +-
.../provider/TieredFileSegment.java | 291 ++++++++++--------
.../provider/TieredStoreProvider.java | 8 +-
.../provider/posix/PosixFileSegment.java | 4 +-
.../CommitLogInputStream.java} | 30 +-
.../FileSegmentInputStream.java} | 49 ++-
.../FileSegmentInputStreamFactory.java} | 26 +-
.../tieredstore/TieredMessageStoreTest.java | 14 +-
.../tieredstore/file/TieredFlatFileTest.java | 2 +
.../tieredstore/file/TieredIndexFileTest.java | 2 +
...m.java => MockFileSegmentInputStream.java} | 8 +-
.../TieredFileSegmentInputStreamTest.java | 24 +-
.../provider/TieredFileSegmentTest.java | 89 +++++-
.../provider/memory/MemoryFileSegment.java | 27 +-
.../memory/MemoryFileSegmentWithoutCheck.java | 4 +-
17 files changed, 427 insertions(+), 237 deletions(-)
rename tieredstore/src/main/java/org/apache/rocketmq/tieredstore/provider/{inputstream/TieredCommitLogInputStream.java => stream/CommitLogInputStream.java} (88%)
rename tieredstore/src/main/java/org/apache/rocketmq/tieredstore/provider/{inputstream/TieredFileSegmentInputStream.java => stream/FileSegmentInputStream.java} (77%)
rename tieredstore/src/main/java/org/apache/rocketmq/tieredstore/provider/{inputstream/TieredFileSegmentInputStreamFactory.java => stream/FileSegmentInputStreamFactory.java} (54%)
rename tieredstore/src/test/java/org/apache/rocketmq/tieredstore/provider/{MockTieredFileSegmentInputStream.java => MockFileSegmentInputStream.java} (82%)
@@ -318,8 +318,7 @@ public class TieredDispatcher extends ServiceThread implements CommitLogDispatch
continue;
case FILE_CLOSED:
tieredFlatFileManager.destroyCompositeFile(flatFile.getMessageQueue());
- logger.info("TieredDispatcher#dispatchFlatFile: file has been close and destroy, " +
- "topic: {}, queueId: {}", topic, queueId);
+ logger.info("File has been closed and destroy, topic: {}, queueId: {}", topic, queueId);
return;
default:
dispatchOffset--;
@@ -273,15 +273,17 @@ public class TieredMessageFetcher implements MessageStoreFetcher {
TieredStoreMetricsManager.cacheHit.add(resultWrapperList.size(), attributes);
}
- // if no cached message found and there is currently an inflight request, wait for the request to end before continuing
+ // If there are no messages in the cache and there are currently requests being pulled.
+ // We need to wait for the request to return before continuing.
if (resultWrapperList.isEmpty() && waitInflightRequest) {
- CompletableFuture<Long> future = flatFile.getInflightRequest(group, queueOffset, maxCount)
- .getFuture(queueOffset);
+ CompletableFuture<Long> future =
+ flatFile.getInflightRequest(group, queueOffset, maxCount).getFuture(queueOffset);
if (!future.isDone()) {
Stopwatch stopwatch = Stopwatch.createStarted();
// to prevent starvation issues, only allow waiting for inflight request once
return future.thenCompose(v -> {
- LOGGER.debug("TieredMessageFetcher#getMessageFromCacheAsync: wait for inflight request cost: {}ms", stopwatch.elapsed(TimeUnit.MILLISECONDS));
+ LOGGER.debug("MessageFetcher#getMessageFromCacheAsync: wait for response cost: {}ms",
+ stopwatch.elapsed(TimeUnit.MILLISECONDS));
return getMessageFromCacheAsync(flatFile, group, queueOffset, maxCount, false);
});
}
@@ -302,7 +304,8 @@ public class TieredMessageFetcher implements MessageStoreFetcher {
// if cache hit, result will be returned immediately and asynchronously prefetch messages for later requests
if (!resultWrapperList.isEmpty()) {
- LOGGER.debug("TieredMessageFetcher#getMessageFromCacheAsync: cache hit: topic: {}, queue: {}, queue offset: {}, max message num: {}, cache hit num: {}",
+ LOGGER.debug("MessageFetcher#getMessageFromCacheAsync: cache hit: " +
+ "topic: {}, queue: {}, queue offset: {}, max message num: {}, cache hit num: {}",
mq.getTopic(), mq.getQueueId(), queueOffset, maxCount, resultWrapperList.size());
prefetchMessage(flatFile, group, maxCount, lastGetOffset + 1);
@@ -316,8 +319,10 @@ public class TieredMessageFetcher implements MessageStoreFetcher {
}
// if cache is miss, immediately pull messages
- LOGGER.warn("TieredMessageFetcher#getMessageFromCacheAsync: cache miss: topic: {}, queue: {}, queue offset: {}, max message num: {}",
+ LOGGER.warn("TieredMessageFetcher#getMessageFromCacheAsync: cache miss: " +
+ "topic: {}, queue: {}, queue offset: {}, max message num: {}",
mq.getTopic(), mq.getQueueId(), queueOffset, maxCount);
+
CompletableFuture<GetMessageResult> resultFuture;
synchronized (flatFile) {
int batchSize = maxCount * storeConfig.getReadAheadMinFactor();
@@ -453,42 +458,42 @@ public class TieredMessageFetcher implements MessageStoreFetcher {
public CompletableFuture<GetMessageResult> getMessageAsync(
String group, String topic, int queueId, long queueOffset, int maxCount, final MessageFilter messageFilter) {
+ GetMessageResult result = new GetMessageResult();
CompositeQueueFlatFile flatFile = flatFileManager.getFlatFile(new MessageQueue(topic, brokerName, queueId));
+
if (flatFile == null) {
- GetMessageResult result = new GetMessageResult();
result.setNextBeginOffset(queueOffset);
result.setStatus(GetMessageStatus.NO_MATCHED_LOGIC_QUEUE);
return CompletableFuture.completedFuture(result);
}
- GetMessageResult result = new GetMessageResult();
- long minQueueOffset = flatFile.getConsumeQueueMinOffset();
- long maxQueueOffset = flatFile.getConsumeQueueCommitOffset();
- result.setMinOffset(minQueueOffset);
- result.setMaxOffset(maxQueueOffset);
+ // Max queue offset means next message put position
+ result.setMinOffset(flatFile.getConsumeQueueMinOffset());
+ result.setMaxOffset(flatFile.getConsumeQueueCommitOffset());
+
+ // Fill result according file offset.
+ // Offset range | Result | Fix to
+ // (-oo, 0] | no message | current offset
+ // (0, min) | too small | min offset
+ // [min, max) | correct |
+ // [max, max] | overflow one | max offset
+ // (max, +oo) | overflow badly | max offset
- if (flatFile.getConsumeQueueCommitOffset() <= 0) {
+ if (result.getMaxOffset() <= 0) {
result.setStatus(GetMessageStatus.NO_MESSAGE_IN_QUEUE);
result.setNextBeginOffset(queueOffset);
return CompletableFuture.completedFuture(result);
- }
-
- // request range | result
- // (0, min) | too small
- // [min, max) | correct
- // [max, max] | overflow one
- // (max, +oo) | overflow badly
- if (queueOffset < minQueueOffset) {
+ } else if (queueOffset < result.getMinOffset()) {
result.setStatus(GetMessageStatus.OFFSET_TOO_SMALL);
- result.setNextBeginOffset(flatFile.getConsumeQueueMinOffset());
+ result.setNextBeginOffset(result.getMinOffset());
return CompletableFuture.completedFuture(result);
- } else if (queueOffset == maxQueueOffset) {
+ } else if (queueOffset == result.getMaxOffset()) {
result.setStatus(GetMessageStatus.OFFSET_OVERFLOW_ONE);
- result.setNextBeginOffset(flatFile.getConsumeQueueCommitOffset());
+ result.setNextBeginOffset(result.getMaxOffset());
return CompletableFuture.completedFuture(result);
- } else if (queueOffset > maxQueueOffset) {
+ } else if (queueOffset > result.getMaxOffset()) {
result.setStatus(GetMessageStatus.OFFSET_OVERFLOW_BADLY);
- result.setNextBeginOffset(flatFile.getConsumeQueueCommitOffset());
+ result.setNextBeginOffset(result.getMaxOffset());
return CompletableFuture.completedFuture(result);
}
@@ -99,11 +99,11 @@ public class TieredMessageStore extends AbstractPluginMessageStore {
return storeConfig;
}
- public boolean viaTieredStorage(String topic, int queueId, long offset) {
- return viaTieredStorage(topic, queueId, offset, 1);
+ public boolean fetchFromCurrentStore(String topic, int queueId, long offset) {
+ return fetchFromCurrentStore(topic, queueId, offset, 1);
}
- public boolean viaTieredStorage(String topic, int queueId, long offset, int batchSize) {
+ public boolean fetchFromCurrentStore(String topic, int queueId, long offset, int batchSize) {
TieredMessageStoreConfig.TieredStorageLevel deepStorageLevel = storeConfig.getTieredStorageLevel();
if (deepStorageLevel.check(TieredMessageStoreConfig.TieredStorageLevel.FORCE)) {
@@ -146,8 +146,10 @@ public class TieredMessageStore extends AbstractPluginMessageStore {
public CompletableFuture<GetMessageResult> getMessageAsync(String group, String topic,
int queueId, long offset, int maxMsgNums, MessageFilter messageFilter) {
- if (!viaTieredStorage(topic, queueId, offset, maxMsgNums)) {
- logger.trace("GetMessageAsync from next store topic: {}, queue: {}, offset: {}", topic, queueId, offset);
+ if (fetchFromCurrentStore(topic, queueId, offset, maxMsgNums)) {
+ logger.trace("GetMessageAsync from current store, topic: {}, queue: {}, offset: {}", topic, queueId, offset);
+ } else {
+ logger.trace("GetMessageAsync from next store, topic: {}, queue: {}, offset: {}", topic, queueId, offset);
return next.getMessageAsync(group, topic, queueId, offset, maxMsgNums, messageFilter);
}
@@ -168,14 +170,14 @@ public class TieredMessageStore extends AbstractPluginMessageStore {
if (next.checkInStoreByConsumeOffset(topic, queueId, offset)) {
TieredStoreMetricsManager.fallbackTotal.add(1, latencyAttributes);
- logger.debug("GetMessageAsync not found then try back to next store, result: {}, " +
+ logger.debug("GetMessageAsync not found, then back to next store, result: {}, " +
"topic: {}, queue: {}, queue offset: {}, offset range: {}-{}",
result.getStatus(), topic, queueId, offset, result.getMinOffset(), result.getMaxOffset());
return next.getMessage(group, topic, queueId, offset, maxMsgNums, messageFilter);
}
}
- // system topic
+ // Fetch system topic data from the broker when using the force level.
if (result.getStatus() == GetMessageStatus.NO_MATCHED_LOGIC_QUEUE) {
if (TieredStoreUtil.isSystemTopic(topic) || PopAckConstants.isStartWithRevivePrefix(topic)) {
return next.getMessage(group, topic, queueId, offset, maxMsgNums, messageFilter);
@@ -198,7 +200,7 @@ public class TieredMessageStore extends AbstractPluginMessageStore {
TieredStoreMetricsManager.messagesOutTotal.add(result.getMessageCount(), messagesOutAttributes);
}
- // fix min or max offset according next store
+ // Fix min or max offset according next store at last
long minOffsetInQueue = next.getMinOffsetInQueue(topic, queueId);
if (minOffsetInQueue >= 0 && minOffsetInQueue < result.getMinOffset()) {
result.setMinOffset(minOffsetInQueue);
@@ -209,7 +211,7 @@ public class TieredMessageStore extends AbstractPluginMessageStore {
}
return result;
}).exceptionally(e -> {
- logger.error("GetMessageAsync from tiered store failed: ", e);
+ logger.error("GetMessageAsync from tiered store failed", e);
return next.getMessage(group, topic, queueId, offset, maxMsgNums, messageFilter);
});
}
@@ -251,7 +253,7 @@ public class TieredMessageStore extends AbstractPluginMessageStore {
.build();
TieredStoreMetricsManager.apiLatency.record(stopwatch.elapsed(TimeUnit.MILLISECONDS), latencyAttributes);
if (time < 0) {
- logger.debug("TieredMessageStore#getEarliestMessageTimeAsync: get earliest message time failed, try to get earliest message time from next store: topic: {}, queue: {}",
+ logger.debug("GetEarliestMessageTimeAsync failed, try to get earliest message time from next store: topic: {}, queue: {}",
topic, queueId);
return finalNextEarliestMessageTime != Long.MAX_VALUE ? finalNextEarliestMessageTime : -1;
}
@@ -262,7 +264,7 @@ public class TieredMessageStore extends AbstractPluginMessageStore {
@Override
public CompletableFuture<Long> getMessageStoreTimeStampAsync(String topic, int queueId,
long consumeQueueOffset) {
- if (viaTieredStorage(topic, queueId, consumeQueueOffset)) {
+ if (fetchFromCurrentStore(topic, queueId, consumeQueueOffset)) {
Stopwatch stopwatch = Stopwatch.createStarted();
return fetcher.getMessageStoreTimeStampAsync(topic, queueId, consumeQueueOffset)
.thenApply(time -> {
@@ -272,7 +274,7 @@ public class TieredMessageStore extends AbstractPluginMessageStore {
.build();
TieredStoreMetricsManager.apiLatency.record(stopwatch.elapsed(TimeUnit.MILLISECONDS), latencyAttributes);
if (time == -1) {
- logger.debug("TieredMessageStore#getMessageStoreTimeStampAsync: get message time failed, try to get message time from next store: topic: {}, queue: {}, queue offset: {}",
+ logger.debug("GetEarliestMessageTimeAsync failed, try to get message time from next store, topic: {}, queue: {}, queue offset: {}",
topic, queueId, consumeQueueOffset);
return next.getMessageStoreTimeStamp(topic, queueId, consumeQueueOffset);
}
@@ -16,14 +16,11 @@
*/
package org.apache.rocketmq.tieredstore.provider;
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Stopwatch;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Semaphore;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
@@ -35,8 +32,8 @@ import org.apache.rocketmq.tieredstore.exception.TieredStoreException;
import org.apache.rocketmq.tieredstore.file.TieredCommitLog;
import org.apache.rocketmq.tieredstore.file.TieredConsumeQueue;
import org.apache.rocketmq.tieredstore.file.TieredIndexFile;
-import org.apache.rocketmq.tieredstore.provider.inputstream.TieredFileSegmentInputStream;
-import org.apache.rocketmq.tieredstore.provider.inputstream.TieredFileSegmentInputStreamFactory;
+import org.apache.rocketmq.tieredstore.provider.stream.FileSegmentInputStream;
+import org.apache.rocketmq.tieredstore.provider.stream.FileSegmentInputStreamFactory;
import org.apache.rocketmq.tieredstore.util.MessageBufferUtil;
import org.apache.rocketmq.tieredstore.util.TieredStoreUtil;
@@ -50,22 +47,23 @@ public abstract class TieredFileSegment implements Comparable<TieredFileSegment>
protected final TieredMessageStoreConfig storeConfig;
private final long maxSize;
- private final ReentrantLock bufferLock;
- private final Semaphore commitLock;
+ private final ReentrantLock bufferLock = new ReentrantLock();
+ private final Semaphore commitLock = new Semaphore(1);
- private volatile boolean full;
- private volatile boolean closed;
+ private volatile boolean full = false;
+ private volatile boolean closed = false;
- private volatile long minTimestamp;
- private volatile long maxTimestamp;
- private volatile long commitPosition;
- private volatile long appendPosition;
+ private volatile long minTimestamp = Long.MAX_VALUE;
+ private volatile long maxTimestamp = Long.MAX_VALUE;
+ private volatile long commitPosition = 0L;
+ private volatile long appendPosition = 0L;
// only used in commitLog
- private volatile long dispatchCommitOffset = 0;
+ private volatile long dispatchCommitOffset = 0L;
private ByteBuffer codaBuffer;
- private List<ByteBuffer> uploadBufferList = new ArrayList<>();
+ private List<ByteBuffer> bufferList = new ArrayList<>();
+ private FileSegmentInputStream fileSegmentInputStream;
private CompletableFuture<Boolean> flightCommitRequest = CompletableFuture.completedFuture(false);
public TieredFileSegment(TieredMessageStoreConfig storeConfig,
@@ -75,21 +73,13 @@ public abstract class TieredFileSegment implements Comparable<TieredFileSegment>
this.fileType = fileType;
this.filePath = filePath;
this.baseOffset = baseOffset;
-
- this.closed = false;
- this.bufferLock = new ReentrantLock();
- this.commitLock = new Semaphore(1);
-
- this.commitPosition = 0L;
- this.appendPosition = 0L;
- this.minTimestamp = Long.MAX_VALUE;
- this.maxTimestamp = Long.MAX_VALUE;
-
- // The max segment size of a file is determined by the file type
- this.maxSize = getMaxSizeAccordingFileType(storeConfig);
+ this.maxSize = getMaxSizeByFileType();
}
- private long getMaxSizeAccordingFileType(TieredMessageStoreConfig storeConfig) {
+ /**
+ * The max segment size of a file is determined by the file type
+ */
+ protected long getMaxSizeByFileType() {
switch (fileType) {
case COMMIT_LOG:
return storeConfig.getTieredStoreCommitLogMaxSize();
@@ -184,39 +174,23 @@ public abstract class TieredFileSegment implements Comparable<TieredFileSegment>
this.appendPosition = pos;
}
- private List<ByteBuffer> rollingUploadBuffer() {
+ private List<ByteBuffer> borrowBuffer() {
bufferLock.lock();
try {
- List<ByteBuffer> tmp = uploadBufferList;
- uploadBufferList = new ArrayList<>();
+ List<ByteBuffer> tmp = bufferList;
+ bufferList = new ArrayList<>();
return tmp;
} finally {
bufferLock.unlock();
}
}
- private void sendBackBuffer(TieredFileSegmentInputStream inputStream) {
- bufferLock.lock();
- try {
- List<ByteBuffer> tmpBufferList = inputStream.getUploadBufferList();
- for (ByteBuffer buffer : tmpBufferList) {
- buffer.rewind();
- }
- tmpBufferList.addAll(uploadBufferList);
- uploadBufferList = tmpBufferList;
- if (inputStream.getCodaBuffer() != null) {
- codaBuffer.rewind();
- }
- } finally {
- bufferLock.unlock();
- }
- }
-
@SuppressWarnings("NonAtomicOperationOnVolatileField")
- public AppendResult append(ByteBuffer byteBuf, long timeStamp) {
+ public AppendResult append(ByteBuffer byteBuf, long timestamp) {
if (closed) {
return AppendResult.FILE_CLOSED;
}
+
bufferLock.lock();
try {
if (full || codaBuffer != null) {
@@ -227,7 +201,8 @@ public abstract class TieredFileSegment implements Comparable<TieredFileSegment>
minTimestamp = byteBuf.getLong(TieredIndexFile.INDEX_FILE_HEADER_BEGIN_TIME_STAMP_POSITION);
maxTimestamp = byteBuf.getLong(TieredIndexFile.INDEX_FILE_HEADER_END_TIME_STAMP_POSITION);
appendPosition += byteBuf.remaining();
- uploadBufferList.add(byteBuf);
+ // IndexFile is large and not change after compaction, no need deep copy
+ bufferList.add(byteBuf);
setFull();
return AppendResult.SUCCESS;
}
@@ -236,23 +211,34 @@ public abstract class TieredFileSegment implements Comparable<TieredFileSegment>
setFull();
return AppendResult.FILE_FULL;
}
- if (uploadBufferList.size() > storeConfig.getTieredStoreGroupCommitCount()
+
+ if (bufferList.size() > storeConfig.getTieredStoreGroupCommitCount()
|| appendPosition - commitPosition > storeConfig.getTieredStoreGroupCommitSize()) {
commitAsync();
}
- if (uploadBufferList.size() > storeConfig.getTieredStoreMaxGroupCommitCount()) {
- logger.debug("TieredFileSegment#append: buffer full: file: {}, upload buffer size: {}",
- getPath(), uploadBufferList.size());
+
+ if (bufferList.size() > storeConfig.getTieredStoreMaxGroupCommitCount()) {
+ logger.debug("File segment append buffer full, file: {}, buffer size: {}, pending bytes: {}",
+ getPath(), bufferList.size(), appendPosition - commitPosition);
return AppendResult.BUFFER_FULL;
}
- if (timeStamp != Long.MAX_VALUE) {
- maxTimestamp = timeStamp;
+
+ if (timestamp != Long.MAX_VALUE) {
+ maxTimestamp = timestamp;
if (minTimestamp == Long.MAX_VALUE) {
- minTimestamp = timeStamp;
+ minTimestamp = timestamp;
}
}
+
appendPosition += byteBuf.remaining();
- uploadBufferList.add(byteBuf);
+
+ // deep copy buffer
+ ByteBuffer byteBuffer = ByteBuffer.allocateDirect(byteBuf.remaining());
+ byteBuffer.put(byteBuf);
+ byteBuffer.flip();
+ byteBuf.rewind();
+
+ bufferList.add(byteBuffer);
return AppendResult.SUCCESS;
} finally {
bufferLock.unlock();
@@ -267,7 +253,6 @@ public abstract class TieredFileSegment implements Comparable<TieredFileSegment>
return appendPosition;
}
- @VisibleForTesting
public void setAppendPosition(long appendPosition) {
this.appendPosition = appendPosition;
}
@@ -333,6 +318,8 @@ public abstract class TieredFileSegment implements Comparable<TieredFileSegment>
if (closed) {
return false;
}
+ // result is false when we send real commit request
+ // use join for wait flight request done
Boolean result = commitAsync().join();
if (!result) {
result = flightCommitRequest.join();
@@ -340,92 +327,156 @@ public abstract class TieredFileSegment implements Comparable<TieredFileSegment>
return result;
}
+ private void releaseCommitLock() {
+ if (commitLock.availablePermits() == 0) {
+ commitLock.release();
+ } else {
+ logger.error("[Bug] FileSegmentCommitAsync, lock is already released: available permits: {}",
+ commitLock.availablePermits());
+ }
+ }
+
+ private void updateDispatchCommitOffset(List<ByteBuffer> bufferList) {
+ if (fileType == FileSegmentType.COMMIT_LOG && bufferList.size() > 0) {
+ dispatchCommitOffset =
+ MessageBufferUtil.getQueueOffset(bufferList.get(bufferList.size() - 1));
+ }
+ }
+
+ /**
+ * @return false: commit, true: no commit operation
+ */
@SuppressWarnings("NonAtomicOperationOnVolatileField")
public CompletableFuture<Boolean> commitAsync() {
if (closed) {
return CompletableFuture.completedFuture(false);
}
- Stopwatch stopwatch = Stopwatch.createStarted();
+
if (!needCommit()) {
return CompletableFuture.completedFuture(true);
}
- try {
- int permits = commitLock.drainPermits();
- if (permits <= 0) {
- return CompletableFuture.completedFuture(false);
- }
- } catch (Exception e) {
+
+ if (commitLock.drainPermits() <= 0) {
return CompletableFuture.completedFuture(false);
}
- List<ByteBuffer> bufferList = rollingUploadBuffer();
- int bufferSize = 0;
- for (ByteBuffer buffer : bufferList) {
- bufferSize += buffer.remaining();
- }
- if (codaBuffer != null) {
- bufferSize += codaBuffer.remaining();
- }
- if (bufferSize == 0) {
- return CompletableFuture.completedFuture(true);
- }
- TieredFileSegmentInputStream inputStream = TieredFileSegmentInputStreamFactory.build(
- fileType, baseOffset + commitPosition, bufferList, codaBuffer, bufferSize);
- int finalBufferSize = bufferSize;
+
try {
- flightCommitRequest = commit0(inputStream, commitPosition, bufferSize, fileType != FileSegmentType.INDEX)
+ if (fileSegmentInputStream != null) {
+ long fileSize = this.getSize();
+ if (fileSize == -1L) {
+ logger.error("Get commit position error before commit, Commit: %d, Expect: %d, Current Max: %d, FileName: %s",
+ commitPosition, commitPosition + fileSegmentInputStream.getContentLength(), appendPosition, getPath());
+ releaseCommitLock();
+ return CompletableFuture.completedFuture(false);
+ } else {
+ if (correctPosition(fileSize, null)) {
+ updateDispatchCommitOffset(fileSegmentInputStream.getBufferList());
+ fileSegmentInputStream = null;
+ }
+ }
+ }
+
+ int bufferSize;
+ if (fileSegmentInputStream != null) {
+ bufferSize = fileSegmentInputStream.available();
+ } else {
+ List<ByteBuffer> bufferList = borrowBuffer();
+ bufferSize = bufferList.stream().mapToInt(ByteBuffer::remaining).sum()
+ + (codaBuffer != null ? codaBuffer.remaining() : 0);
+ if (bufferSize == 0) {
+ releaseCommitLock();
+ return CompletableFuture.completedFuture(true);
+ }
+ fileSegmentInputStream = FileSegmentInputStreamFactory.build(
+ fileType, baseOffset + commitPosition, bufferList, codaBuffer, bufferSize);
+ }
+
+ return flightCommitRequest = this
+ .commit0(fileSegmentInputStream, commitPosition, bufferSize, fileType != FileSegmentType.INDEX)
.thenApply(result -> {
if (result) {
- if (fileType == FileSegmentType.COMMIT_LOG && bufferList.size() > 0) {
- dispatchCommitOffset = MessageBufferUtil.getQueueOffset(bufferList.get(bufferList.size() - 1));
- }
- commitPosition += finalBufferSize;
+ updateDispatchCommitOffset(fileSegmentInputStream.getBufferList());
+ commitPosition += bufferSize;
+ fileSegmentInputStream = null;
return true;
- }
- sendBackBuffer(inputStream);
- return false;
- })
- .exceptionally(e -> handleCommitException(inputStream, e))
- .whenComplete((result, e) -> {
- if (commitLock.availablePermits() == 0) {
- logger.debug("TieredFileSegment#commitAsync: commit cost: {}ms, file: {}, item count: {}, buffer size: {}", stopwatch.elapsed(TimeUnit.MILLISECONDS), getPath(), bufferList.size(), finalBufferSize);
- commitLock.release();
} else {
- logger.error("[Bug]TieredFileSegment#commitAsync: commit lock is already released: available permits: {}", commitLock.availablePermits());
+ fileSegmentInputStream.rewind();
+ return false;
}
- });
- return flightCommitRequest;
+ })
+ .exceptionally(this::handleCommitException)
+ .whenComplete((result, e) -> releaseCommitLock());
+
} catch (Exception e) {
- handleCommitException(inputStream, e);
- if (commitLock.availablePermits() == 0) {
- logger.debug("TieredFileSegment#commitAsync: commit cost: {}ms, file: {}, item count: {}, buffer size: {}", stopwatch.elapsed(TimeUnit.MILLISECONDS), getPath(), bufferList.size(), finalBufferSize);
- commitLock.release();
- } else {
- logger.error("[Bug]TieredFileSegment#commitAsync: commit lock is already released: available permits: {}", commitLock.availablePermits());
- }
+ handleCommitException(e);
+ releaseCommitLock();
}
return CompletableFuture.completedFuture(false);
}
- private boolean handleCommitException(TieredFileSegmentInputStream inputStream, Throwable e) {
+ private long getCorrectFileSize(Throwable throwable) {
+ if (throwable instanceof TieredStoreException) {
+ long fileSize = ((TieredStoreException) throwable).getPosition();
+ if (fileSize > 0) {
+ return fileSize;
+ }
+ }
+ return getSize();
+ }
+
+ private boolean handleCommitException(Throwable e) {
+ // Get root cause here
Throwable cause = e.getCause() != null ? e.getCause() : e;
- sendBackBuffer(inputStream);
- long realSize = 0;
- if (cause instanceof TieredStoreException && ((TieredStoreException) cause).getPosition() > 0) {
- realSize = ((TieredStoreException) cause).getPosition();
+ long fileSize = this.getCorrectFileSize(cause);
+
+ if (fileSize == -1L) {
+ logger.error("Get commit position error, Commit: %d, Expect: %d, Current Max: %d, FileName: %s",
+ commitPosition, commitPosition + fileSegmentInputStream.getContentLength(), appendPosition, getPath());
+ fileSegmentInputStream.rewind();
+ return false;
+ }
+
+ if (correctPosition(fileSize, cause)) {
+ updateDispatchCommitOffset(fileSegmentInputStream.getBufferList());
+ fileSegmentInputStream = null;
+ return true;
+ } else {
+ fileSegmentInputStream.rewind();
+ return false;
}
- if (realSize <= 0) {
- realSize = getSize();
+ }
+
+ /**
+ * return true to clear buffer
+ */
+ private boolean correctPosition(long fileSize, Throwable throwable) {
+
+ // Current we have three offsets here: commit offset, expect offset, file size.
+ // We guarantee that the commit offset is less than or equal to the expect offset.
+ // Max offset will increase because we can continuously put in new buffers
+ String handleInfo = throwable == null ? "before commit" : "after commit";
+ long expectPosition = commitPosition + fileSegmentInputStream.getContentLength();
+
+ String offsetInfo = String.format("Correct Commit Position, %s, result=[{}], " +
+ "Commit: %d, Expect: %d, Current Max: %d, FileSize: %d, FileName: %s",
+ handleInfo, commitPosition, expectPosition, appendPosition, fileSize, this.getPath());
+
+ // We are believing that the file size returned by the server is correct,
+ // can reset the commit offset to the file size reported by the storage system.
+ if (fileSize == expectPosition) {
+ logger.info(offsetInfo, "Success", throwable);
+ commitPosition = fileSize;
+ return true;
}
- if (realSize > 0 && realSize > commitPosition) {
- logger.error("TieredFileSegment#handleCommitException: commit failed: file: {}, try to fix position: origin: {}, real: {}", getPath(), commitPosition, realSize, cause);
- // TODO check if this diff part is uploaded to backend storage
- long diff = appendPosition - commitPosition;
- commitPosition = realSize;
- appendPosition = realSize + diff;
- // TODO check if appendPosition is large than maxOffset
- } else if (realSize < commitPosition) {
- logger.error("[Bug]TieredFileSegment#handleCommitException: commit failed: file: {}, can not fix position: origin: {}, real: {}", getPath(), commitPosition, realSize, cause);
+
+ if (fileSize < commitPosition) {
+ logger.error(offsetInfo, "FileSizeIncorrect", throwable);
+ } else if (fileSize == commitPosition) {
+ logger.warn(offsetInfo, "CommitFailed", throwable);
+ } else if (fileSize > commitPosition) {
+ logger.warn(offsetInfo, "PartialSuccess", throwable);
}
+ commitPosition = fileSize;
return false;
}
}
@@ -18,7 +18,7 @@ package org.apache.rocketmq.tieredstore.provider;
import java.nio.ByteBuffer;
import java.util.concurrent.CompletableFuture;
-import org.apache.rocketmq.tieredstore.provider.inputstream.TieredFileSegmentInputStream;
+import org.apache.rocketmq.tieredstore.provider.stream.FileSegmentInputStream;
public interface TieredStoreProvider {
@@ -30,7 +30,9 @@ public interface TieredStoreProvider {
String getPath();
/**
- * Get file size in backend file system
+ * Get the real length of the file.
+ * Return 0 if the file does not exist,
+ * Return -1 if system get size failed.
*
* @return file real size
*/
@@ -71,5 +73,5 @@ public interface TieredStoreProvider {
* @param append try to append or create a new file
* @return put result, <code>true</code> if data successfully write; <code>false</code> otherwise
*/
- CompletableFuture<Boolean> commit0(TieredFileSegmentInputStream inputStream,long position, int length, boolean append);
+ CompletableFuture<Boolean> commit0(FileSegmentInputStream inputStream,long position, int length, boolean append);
}
@@ -36,7 +36,7 @@ import org.apache.rocketmq.tieredstore.common.TieredMessageStoreConfig;
import org.apache.rocketmq.tieredstore.common.TieredStoreExecutor;
import org.apache.rocketmq.tieredstore.metrics.TieredStoreMetricsManager;
import org.apache.rocketmq.tieredstore.provider.TieredFileSegment;
-import org.apache.rocketmq.tieredstore.provider.inputstream.TieredFileSegmentInputStream;
+import org.apache.rocketmq.tieredstore.provider.stream.FileSegmentInputStream;
import org.apache.rocketmq.tieredstore.util.TieredStoreUtil;
import static org.apache.rocketmq.tieredstore.metrics.TieredStoreMetricsConstant.LABEL_FILE_TYPE;
@@ -184,7 +184,7 @@ public class PosixFileSegment extends TieredFileSegment {
@Override
public CompletableFuture<Boolean> commit0(
- TieredFileSegmentInputStream inputStream, long position, int length, boolean append) {
+ FileSegmentInputStream inputStream, long position, int length, boolean append) {
Stopwatch stopwatch = Stopwatch.createStarted();
AttributesBuilder attributesBuilder = newAttributesBuilder()
similarity index 88%
rename from tieredstore/src/main/java/org/apache/rocketmq/tieredstore/provider/inputstream/TieredCommitLogInputStream.java
rename to tieredstore/src/main/java/org/apache/rocketmq/tieredstore/provider/stream/CommitLogInputStream.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.rocketmq.tieredstore.provider.inputstream;
+package org.apache.rocketmq.tieredstore.provider.stream;
import java.io.IOException;
import java.nio.ByteBuffer;
@@ -23,20 +23,23 @@ import java.util.List;
import org.apache.rocketmq.tieredstore.common.FileSegmentType;
import org.apache.rocketmq.tieredstore.util.MessageBufferUtil;
-public class TieredCommitLogInputStream extends TieredFileSegmentInputStream {
+public class CommitLogInputStream extends FileSegmentInputStream {
/**
* commitLogOffset is the real physical offset of the commitLog buffer which is being read
*/
+ private final long startCommitLogOffset;
+
private long commitLogOffset;
private final ByteBuffer codaBuffer;
private long markCommitLogOffset = -1;
- public TieredCommitLogInputStream(FileSegmentType fileType, long startOffset,
+ public CommitLogInputStream(FileSegmentType fileType, long startOffset,
List<ByteBuffer> uploadBufferList, ByteBuffer codaBuffer, int contentLength) {
super(fileType, uploadBufferList, contentLength);
+ this.startCommitLogOffset = startOffset;
this.commitLogOffset = startOffset;
this.codaBuffer = codaBuffer;
}
@@ -53,6 +56,15 @@ public class TieredCommitLogInputStream extends TieredFileSegmentInputStream {
this.commitLogOffset = markCommitLogOffset;
}
+ @Override
+ public synchronized void rewind() {
+ super.rewind();
+ this.commitLogOffset = this.startCommitLogOffset;
+ if (this.codaBuffer != null) {
+ this.codaBuffer.rewind();
+ }
+ }
+
@Override
public ByteBuffer getCodaBuffer() {
return this.codaBuffer;
@@ -64,17 +76,17 @@ public class TieredCommitLogInputStream extends TieredFileSegmentInputStream {
return -1;
}
readPosition++;
- if (curReadBufferIndex >= uploadBufferList.size()) {
+ if (curReadBufferIndex >= bufferList.size()) {
return readCoda();
}
int res;
if (readPosInCurBuffer >= curBuffer.remaining()) {
curReadBufferIndex++;
- if (curReadBufferIndex >= uploadBufferList.size()) {
+ if (curReadBufferIndex >= bufferList.size()) {
readPosInCurBuffer = 0;
return readCoda();
}
- curBuffer = uploadBufferList.get(curReadBufferIndex);
+ curBuffer = bufferList.get(curReadBufferIndex);
commitLogOffset += readPosInCurBuffer;
readPosInCurBuffer = 0;
}
@@ -119,9 +131,9 @@ public class TieredCommitLogInputStream extends TieredFileSegmentInputStream {
int posInCurBuffer = readPosInCurBuffer;
long curCommitLogOffset = commitLogOffset;
ByteBuffer curBuf = curBuffer;
- while (needRead > 0 && bufIndex <= uploadBufferList.size()) {
+ while (needRead > 0 && bufIndex <= bufferList.size()) {
int readLen, remaining, realReadLen = 0;
- if (bufIndex == uploadBufferList.size()) {
+ if (bufIndex == bufferList.size()) {
// read from coda buffer
remaining = codaBuffer.remaining() - posInCurBuffer;
readLen = Math.min(remaining, needRead);
@@ -137,7 +149,7 @@ public class TieredCommitLogInputStream extends TieredFileSegmentInputStream {
}
remaining = curBuf.remaining() - posInCurBuffer;
readLen = Math.min(remaining, needRead);
- curBuf = uploadBufferList.get(bufIndex);
+ curBuf = bufferList.get(bufIndex);
if (posInCurBuffer < MessageBufferUtil.PHYSICAL_OFFSET_POSITION) {
realReadLen = Math.min(MessageBufferUtil.PHYSICAL_OFFSET_POSITION - posInCurBuffer, readLen);
// read from commitLog buffer
similarity index 77%
rename from tieredstore/src/main/java/org/apache/rocketmq/tieredstore/provider/inputstream/TieredFileSegmentInputStream.java
rename to tieredstore/src/main/java/org/apache/rocketmq/tieredstore/provider/stream/FileSegmentInputStream.java
@@ -15,15 +15,16 @@
* limitations under the License.
*/
-package org.apache.rocketmq.tieredstore.provider.inputstream;
+package org.apache.rocketmq.tieredstore.provider.stream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.util.List;
+import org.apache.commons.collections.CollectionUtils;
import org.apache.rocketmq.tieredstore.common.FileSegmentType;
-public class TieredFileSegmentInputStream extends InputStream {
+public class FileSegmentInputStream extends InputStream {
/**
* file type, can be commitlog, consume queue or indexfile now
@@ -33,7 +34,7 @@ public class TieredFileSegmentInputStream extends InputStream {
/**
* hold bytebuffer
*/
- protected final List<ByteBuffer> uploadBufferList;
+ protected final List<ByteBuffer> bufferList;
/**
* total remaining of bytebuffer list
@@ -65,13 +66,13 @@ public class TieredFileSegmentInputStream extends InputStream {
private int markReadPosInCurBuffer = -1;
- public TieredFileSegmentInputStream(FileSegmentType fileType, List<ByteBuffer> uploadBufferList,
- int contentLength) {
+ public FileSegmentInputStream(
+ FileSegmentType fileType, List<ByteBuffer> bufferList, int contentLength) {
this.fileType = fileType;
this.contentLength = contentLength;
- this.uploadBufferList = uploadBufferList;
- if (uploadBufferList != null && uploadBufferList.size() > 0) {
- this.curBuffer = uploadBufferList.get(curReadBufferIndex);
+ this.bufferList = bufferList;
+ if (bufferList != null && bufferList.size() > 0) {
+ this.curBuffer = bufferList.get(curReadBufferIndex);
}
}
@@ -95,18 +96,34 @@ public class TieredFileSegmentInputStream extends InputStream {
this.readPosition = markReadPosition;
this.curReadBufferIndex = markCurReadBufferIndex;
this.readPosInCurBuffer = markReadPosInCurBuffer;
- if (this.curReadBufferIndex < uploadBufferList.size()) {
- this.curBuffer = uploadBufferList.get(curReadBufferIndex);
+ if (this.curReadBufferIndex < bufferList.size()) {
+ this.curBuffer = bufferList.get(curReadBufferIndex);
}
}
+ public synchronized void rewind() {
+ this.readPosition = 0;
+ this.curReadBufferIndex = 0;
+ this.readPosInCurBuffer = 0;
+ if (CollectionUtils.isNotEmpty(bufferList)) {
+ this.curBuffer = bufferList.get(0);
+ for (ByteBuffer buffer : bufferList) {
+ buffer.rewind();
+ }
+ }
+ }
+
+ public int getContentLength() {
+ return contentLength;
+ }
+
@Override
public int available() {
return contentLength - readPosition;
}
- public List<ByteBuffer> getUploadBufferList() {
- return uploadBufferList;
+ public List<ByteBuffer> getBufferList() {
+ return bufferList;
}
public ByteBuffer getCodaBuffer() {
@@ -121,10 +138,10 @@ public class TieredFileSegmentInputStream extends InputStream {
readPosition++;
if (readPosInCurBuffer >= curBuffer.remaining()) {
curReadBufferIndex++;
- if (curReadBufferIndex >= uploadBufferList.size()) {
+ if (curReadBufferIndex >= bufferList.size()) {
return -1;
}
- curBuffer = uploadBufferList.get(curReadBufferIndex);
+ curBuffer = bufferList.get(curReadBufferIndex);
readPosInCurBuffer = 0;
}
return curBuffer.get(readPosInCurBuffer++) & 0xff;
@@ -153,8 +170,8 @@ public class TieredFileSegmentInputStream extends InputStream {
int bufIndex = curReadBufferIndex;
int posInCurBuffer = readPosInCurBuffer;
ByteBuffer curBuf = curBuffer;
- while (needRead > 0 && bufIndex < uploadBufferList.size()) {
- curBuf = uploadBufferList.get(bufIndex);
+ while (needRead > 0 && bufIndex < bufferList.size()) {
+ curBuf = bufferList.get(bufIndex);
int remaining = curBuf.remaining() - posInCurBuffer;
int readLen = Math.min(remaining, needRead);
// read from curBuf
similarity index 54%
rename from tieredstore/src/main/java/org/apache/rocketmq/tieredstore/provider/inputstream/TieredFileSegmentInputStreamFactory.java
rename to tieredstore/src/main/java/org/apache/rocketmq/tieredstore/provider/stream/FileSegmentInputStreamFactory.java
@@ -15,30 +15,34 @@
* limitations under the License.
*/
-package org.apache.rocketmq.tieredstore.provider.inputstream;
+package org.apache.rocketmq.tieredstore.provider.stream;
import java.nio.ByteBuffer;
import java.util.List;
import org.apache.rocketmq.tieredstore.common.FileSegmentType;
-public class TieredFileSegmentInputStreamFactory {
+public class FileSegmentInputStreamFactory {
- public static TieredFileSegmentInputStream build(FileSegmentType fileType,
- long startOffset, List<ByteBuffer> uploadBufferList, ByteBuffer codaBuffer, int contentLength) {
+ public static FileSegmentInputStream build(
+ FileSegmentType fileType, long offset, List<ByteBuffer> bufferList, ByteBuffer byteBuffer, int length) {
+
+ if (bufferList == null) {
+ throw new IllegalArgumentException("bufferList is null");
+ }
switch (fileType) {
case COMMIT_LOG:
- return new TieredCommitLogInputStream(
- fileType, startOffset, uploadBufferList, codaBuffer, contentLength);
+ return new CommitLogInputStream(
+ fileType, offset, bufferList, byteBuffer, length);
case CONSUME_QUEUE:
- return new TieredFileSegmentInputStream(fileType, uploadBufferList, contentLength);
+ return new FileSegmentInputStream(fileType, bufferList, length);
case INDEX:
- if (uploadBufferList.size() != 1) {
- throw new IllegalArgumentException("uploadBufferList size in INDEX type input stream must be 1");
+ if (bufferList.size() != 1) {
+ throw new IllegalArgumentException("buffer block size must be 1 when file type is IndexFile");
}
- return new TieredFileSegmentInputStream(fileType, uploadBufferList, contentLength);
+ return new FileSegmentInputStream(fileType, bufferList, length);
default:
- throw new IllegalArgumentException("fileType is not supported");
+ throw new IllegalArgumentException("file type is not supported");
}
}
}
@@ -130,36 +130,36 @@ public class TieredMessageStoreTest {
// TieredStorageLevel.DISABLE
properties.setProperty("tieredStorageLevel", "0");
configuration.update(properties);
- Assert.assertFalse(store.viaTieredStorage(mq.getTopic(), mq.getQueueId(), 0));
+ Assert.assertFalse(store.fetchFromCurrentStore(mq.getTopic(), mq.getQueueId(), 0));
// TieredStorageLevel.NOT_IN_DISK
properties.setProperty("tieredStorageLevel", "1");
configuration.update(properties);
when(nextStore.checkInStoreByConsumeOffset(anyString(), anyInt(), anyLong())).thenReturn(false);
- Assert.assertTrue(store.viaTieredStorage(mq.getTopic(), mq.getQueueId(), 0));
+ Assert.assertTrue(store.fetchFromCurrentStore(mq.getTopic(), mq.getQueueId(), 0));
when(nextStore.checkInStoreByConsumeOffset(anyString(), anyInt(), anyLong())).thenReturn(true);
- Assert.assertFalse(store.viaTieredStorage(mq.getTopic(), mq.getQueueId(), 0));
+ Assert.assertFalse(store.fetchFromCurrentStore(mq.getTopic(), mq.getQueueId(), 0));
// TieredStorageLevel.NOT_IN_MEM
properties.setProperty("tieredStorageLevel", "2");
configuration.update(properties);
Mockito.when(nextStore.checkInStoreByConsumeOffset(anyString(), anyInt(), anyLong())).thenReturn(false);
Mockito.when(nextStore.checkInMemByConsumeOffset(anyString(), anyInt(), anyLong(), anyInt())).thenReturn(true);
- Assert.assertTrue(store.viaTieredStorage(mq.getTopic(), mq.getQueueId(), 0));
+ Assert.assertTrue(store.fetchFromCurrentStore(mq.getTopic(), mq.getQueueId(), 0));
Mockito.when(nextStore.checkInStoreByConsumeOffset(anyString(), anyInt(), anyLong())).thenReturn(true);
Mockito.when(nextStore.checkInMemByConsumeOffset(anyString(), anyInt(), anyLong(), anyInt())).thenReturn(false);
- Assert.assertTrue(store.viaTieredStorage(mq.getTopic(), mq.getQueueId(), 0));
+ Assert.assertTrue(store.fetchFromCurrentStore(mq.getTopic(), mq.getQueueId(), 0));
Mockito.when(nextStore.checkInStoreByConsumeOffset(anyString(), anyInt(), anyLong())).thenReturn(true);
Mockito.when(nextStore.checkInMemByConsumeOffset(anyString(), anyInt(), anyLong(), anyInt())).thenReturn(true);
- Assert.assertFalse(store.viaTieredStorage(mq.getTopic(), mq.getQueueId(), 0));
+ Assert.assertFalse(store.fetchFromCurrentStore(mq.getTopic(), mq.getQueueId(), 0));
// TieredStorageLevel.FORCE
properties.setProperty("tieredStorageLevel", "3");
configuration.update(properties);
- Assert.assertTrue(store.viaTieredStorage(mq.getTopic(), mq.getQueueId(), 0));
+ Assert.assertTrue(store.fetchFromCurrentStore(mq.getTopic(), mq.getQueueId(), 0));
}
@Test
@@ -24,6 +24,7 @@ import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.tieredstore.TieredStoreTestUtil;
import org.apache.rocketmq.tieredstore.common.FileSegmentType;
import org.apache.rocketmq.tieredstore.common.TieredMessageStoreConfig;
+import org.apache.rocketmq.tieredstore.common.TieredStoreExecutor;
import org.apache.rocketmq.tieredstore.metadata.FileSegmentMetadata;
import org.apache.rocketmq.tieredstore.metadata.TieredMetadataStore;
import org.apache.rocketmq.tieredstore.provider.TieredFileSegment;
@@ -55,6 +56,7 @@ public class TieredFlatFileTest {
public void tearDown() throws IOException {
TieredStoreTestUtil.destroyMetadataStore();
TieredStoreTestUtil.destroyTempDir(storePath);
+ TieredStoreExecutor.shutdown();
}
private List<FileSegmentMetadata> getSegmentMetadataList(TieredMetadataStore metadataStore) {
@@ -87,5 +87,7 @@ public class TieredIndexFileTest {
indexList = indexFile.queryAsync(mq.getTopic(), "key1", 1000, 1200).join();
Assert.assertEquals(1, indexList.size());
+
+ indexFile.destroy();
}
}
similarity index 82%
rename from tieredstore/src/test/java/org/apache/rocketmq/tieredstore/provider/MockTieredFileSegmentInputStream.java
rename to tieredstore/src/test/java/org/apache/rocketmq/tieredstore/provider/MockFileSegmentInputStream.java
@@ -20,13 +20,13 @@ package org.apache.rocketmq.tieredstore.provider;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.util.List;
-import org.apache.rocketmq.tieredstore.provider.inputstream.TieredFileSegmentInputStream;
+import org.apache.rocketmq.tieredstore.provider.stream.FileSegmentInputStream;
-public class MockTieredFileSegmentInputStream extends TieredFileSegmentInputStream {
+public class MockFileSegmentInputStream extends FileSegmentInputStream {
private final InputStream inputStream;
- public MockTieredFileSegmentInputStream(InputStream inputStream) {
+ public MockFileSegmentInputStream(InputStream inputStream) {
super(null, null, Integer.MAX_VALUE);
this.inputStream = inputStream;
}
@@ -43,7 +43,7 @@ public class MockTieredFileSegmentInputStream extends TieredFileSegmentInputStre
}
@Override
- public List<ByteBuffer> getUploadBufferList() {
+ public List<ByteBuffer> getBufferList() {
return null;
}
@@ -28,8 +28,8 @@ import java.util.Random;
import org.apache.rocketmq.tieredstore.common.FileSegmentType;
import org.apache.rocketmq.tieredstore.file.TieredCommitLog;
import org.apache.rocketmq.tieredstore.file.TieredConsumeQueue;
-import org.apache.rocketmq.tieredstore.provider.inputstream.TieredFileSegmentInputStream;
-import org.apache.rocketmq.tieredstore.provider.inputstream.TieredFileSegmentInputStreamFactory;
+import org.apache.rocketmq.tieredstore.provider.stream.FileSegmentInputStream;
+import org.apache.rocketmq.tieredstore.provider.stream.FileSegmentInputStreamFactory;
import org.apache.rocketmq.tieredstore.util.MessageBufferUtil;
import org.apache.rocketmq.tieredstore.util.MessageBufferUtilTest;
import org.junit.Assert;
@@ -57,7 +57,7 @@ public class TieredFileSegmentInputStreamTest {
bufferSize += byteBuffer.remaining();
}
- // build expected byte buffer for verifying the TieredFileSegmentInputStream
+ // build expected byte buffer for verifying the FileSegmentInputStream
ByteBuffer expectedByteBuffer = ByteBuffer.allocate(bufferSize);
for (ByteBuffer byteBuffer : uploadBufferList) {
expectedByteBuffer.put(byteBuffer);
@@ -74,7 +74,7 @@ public class TieredFileSegmentInputStreamTest {
int[] batchReadSizeTestSet = {
MessageBufferUtil.PHYSICAL_OFFSET_POSITION - 1, MessageBufferUtil.PHYSICAL_OFFSET_POSITION, MessageBufferUtil.PHYSICAL_OFFSET_POSITION + 1, MSG_LEN - 1, MSG_LEN, MSG_LEN + 1
};
- verifyReadAndReset(expectedByteBuffer, () -> TieredFileSegmentInputStreamFactory.build(
+ verifyReadAndReset(expectedByteBuffer, () -> FileSegmentInputStreamFactory.build(
FileSegmentType.COMMIT_LOG, COMMIT_LOG_START_OFFSET, uploadBufferList, null, finalBufferSize), finalBufferSize, batchReadSizeTestSet);
}
@@ -98,7 +98,7 @@ public class TieredFileSegmentInputStreamTest {
int codaBufferSize = codaBuffer.remaining();
bufferSize += codaBufferSize;
- // build expected byte buffer for verifying the TieredFileSegmentInputStream
+ // build expected byte buffer for verifying the FileSegmentInputStream
ByteBuffer expectedByteBuffer = ByteBuffer.allocate(bufferSize);
for (ByteBuffer byteBuffer : uploadBufferList) {
expectedByteBuffer.put(byteBuffer);
@@ -119,7 +119,7 @@ public class TieredFileSegmentInputStreamTest {
MSG_LEN - 1, MSG_LEN, MSG_LEN + 1,
bufferSize - 1, bufferSize, bufferSize + 1
};
- verifyReadAndReset(expectedByteBuffer, () -> TieredFileSegmentInputStreamFactory.build(
+ verifyReadAndReset(expectedByteBuffer, () -> FileSegmentInputStreamFactory.build(
FileSegmentType.COMMIT_LOG, COMMIT_LOG_START_OFFSET, uploadBufferList, codaBuffer, finalBufferSize), finalBufferSize, batchReadSizeTestSet);
}
@@ -134,7 +134,7 @@ public class TieredFileSegmentInputStreamTest {
bufferSize += byteBuffer.remaining();
}
- // build expected byte buffer for verifying the TieredFileSegmentInputStream
+ // build expected byte buffer for verifying the FileSegmentInputStream
ByteBuffer expectedByteBuffer = ByteBuffer.allocate(bufferSize);
for (ByteBuffer byteBuffer : uploadBufferList) {
expectedByteBuffer.put(byteBuffer);
@@ -143,7 +143,7 @@ public class TieredFileSegmentInputStreamTest {
int finalBufferSize = bufferSize;
int[] batchReadSizeTestSet = {TieredConsumeQueue.CONSUME_QUEUE_STORE_UNIT_SIZE - 1, TieredConsumeQueue.CONSUME_QUEUE_STORE_UNIT_SIZE, TieredConsumeQueue.CONSUME_QUEUE_STORE_UNIT_SIZE + 1};
- verifyReadAndReset(expectedByteBuffer, () -> TieredFileSegmentInputStreamFactory.build(
+ verifyReadAndReset(expectedByteBuffer, () -> FileSegmentInputStreamFactory.build(
FileSegmentType.CONSUME_QUEUE, COMMIT_LOG_START_OFFSET, uploadBufferList, null, finalBufferSize), bufferSize, batchReadSizeTestSet);
}
@@ -156,16 +156,16 @@ public class TieredFileSegmentInputStreamTest {
byteBuffer.flip();
List<ByteBuffer> uploadBufferList = Arrays.asList(byteBuffer);
- // build expected byte buffer for verifying the TieredFileSegmentInputStream
+ // build expected byte buffer for verifying the FileSegmentInputStream
ByteBuffer expectedByteBuffer = byteBuffer.slice();
- verifyReadAndReset(expectedByteBuffer, () -> TieredFileSegmentInputStreamFactory.build(
+ verifyReadAndReset(expectedByteBuffer, () -> FileSegmentInputStreamFactory.build(
FileSegmentType.INDEX, COMMIT_LOG_START_OFFSET, uploadBufferList, null, byteBuffer.limit()), byteBuffer.limit(), new int[] {23, 24, 25});
}
- private void verifyReadAndReset(ByteBuffer expectedByteBuffer, Supplier<TieredFileSegmentInputStream> constructor,
+ private void verifyReadAndReset(ByteBuffer expectedByteBuffer, Supplier<FileSegmentInputStream> constructor,
int bufferSize, int[] readBatchSizeTestSet) {
- TieredFileSegmentInputStream inputStream = constructor.get();
+ FileSegmentInputStream inputStream = constructor.get();
// verify
verifyInputStream(inputStream, expectedByteBuffer);
@@ -116,13 +116,22 @@ public class TieredFileSegmentTest {
}
@Test
- public void testCommitFailed() {
+ public void testCommitFailedThenSuccess() {
long startTime = System.currentTimeMillis();
MemoryFileSegment segment = (MemoryFileSegment) createFileSegment(FileSegmentType.COMMIT_LOG);
long lastSize = segment.getSize();
- segment.append(MessageBufferUtilTest.buildMockedMessageBuffer(), 0);
- segment.append(MessageBufferUtilTest.buildMockedMessageBuffer(), 0);
-
+ segment.setCheckSize(false);
+ segment.initPosition(lastSize);
+ segment.setSize((int) lastSize);
+
+ ByteBuffer buffer1 = MessageBufferUtilTest.buildMockedMessageBuffer().putLong(
+ MessageBufferUtil.PHYSICAL_OFFSET_POSITION, baseOffset + lastSize);
+ ByteBuffer buffer2 = MessageBufferUtilTest.buildMockedMessageBuffer().putLong(
+ MessageBufferUtil.PHYSICAL_OFFSET_POSITION, baseOffset + lastSize + MessageBufferUtilTest.MSG_LEN);
+ segment.append(buffer1, 0);
+ segment.append(buffer2, 0);
+
+ // Mock new message arrive
segment.blocker = new CompletableFuture<>();
new Thread(() -> {
try {
@@ -131,20 +140,88 @@ public class TieredFileSegmentTest {
Assert.fail(e.getMessage());
}
ByteBuffer buffer = MessageBufferUtilTest.buildMockedMessageBuffer();
+ buffer.putLong(MessageBufferUtil.PHYSICAL_OFFSET_POSITION, MessageBufferUtilTest.MSG_LEN * 2);
buffer.putLong(MessageBufferUtil.STORE_TIMESTAMP_POSITION, startTime);
segment.append(buffer, 0);
segment.blocker.complete(false);
}).start();
+ // Commit failed
segment.commit();
segment.blocker.join();
+ segment.blocker = null;
+
+ // Copy data and assume commit success
+ segment.getMemStore().put(buffer1);
+ segment.getMemStore().put(buffer2);
+ segment.setSize((int) (lastSize + MessageBufferUtilTest.MSG_LEN * 2));
- segment.blocker = new CompletableFuture<>();
- segment.blocker.complete(true);
segment.commit();
+ Assert.assertEquals(lastSize + MessageBufferUtilTest.MSG_LEN * 3, segment.getCommitPosition());
+ Assert.assertEquals(baseOffset + lastSize + MessageBufferUtilTest.MSG_LEN * 3, segment.getCommitOffset());
+ Assert.assertEquals(baseOffset + lastSize + MessageBufferUtilTest.MSG_LEN * 3, segment.getMaxOffset());
+
+ ByteBuffer msg1 = segment.read(lastSize, MessageBufferUtilTest.MSG_LEN);
+ Assert.assertEquals(baseOffset + lastSize, MessageBufferUtil.getCommitLogOffset(msg1));
+
+ ByteBuffer msg2 = segment.read(lastSize + MessageBufferUtilTest.MSG_LEN, MessageBufferUtilTest.MSG_LEN);
+ Assert.assertEquals(baseOffset + lastSize + MessageBufferUtilTest.MSG_LEN, MessageBufferUtil.getCommitLogOffset(msg2));
+
+ ByteBuffer msg3 = segment.read(lastSize + MessageBufferUtilTest.MSG_LEN * 2, MessageBufferUtilTest.MSG_LEN);
+ Assert.assertEquals(baseOffset + lastSize + MessageBufferUtilTest.MSG_LEN * 2, MessageBufferUtil.getCommitLogOffset(msg3));
+ }
+
+ @Test
+ public void testCommitFailed3Times() {
+ long startTime = System.currentTimeMillis();
+ MemoryFileSegment segment = (MemoryFileSegment) createFileSegment(FileSegmentType.COMMIT_LOG);
+ long lastSize = segment.getSize();
+ segment.setCheckSize(false);
+ segment.initPosition(lastSize);
+ segment.setSize((int) lastSize);
+
+ ByteBuffer buffer1 = MessageBufferUtilTest.buildMockedMessageBuffer().putLong(
+ MessageBufferUtil.PHYSICAL_OFFSET_POSITION, baseOffset + lastSize);
+ ByteBuffer buffer2 = MessageBufferUtilTest.buildMockedMessageBuffer().putLong(
+ MessageBufferUtil.PHYSICAL_OFFSET_POSITION, baseOffset + lastSize + MessageBufferUtilTest.MSG_LEN);
+ segment.append(buffer1, 0);
+ segment.append(buffer2, 0);
+
+ // Mock new message arrive
+ segment.blocker = new CompletableFuture<>();
+ new Thread(() -> {
+ try {
+ Thread.sleep(3000);
+ } catch (InterruptedException e) {
+ Assert.fail(e.getMessage());
+ }
+ ByteBuffer buffer = MessageBufferUtilTest.buildMockedMessageBuffer();
+ buffer.putLong(MessageBufferUtil.PHYSICAL_OFFSET_POSITION, MessageBufferUtilTest.MSG_LEN * 2);
+ buffer.putLong(MessageBufferUtil.STORE_TIMESTAMP_POSITION, startTime);
+ segment.append(buffer, 0);
+ segment.blocker.complete(false);
+ }).start();
+
+ for (int i = 0; i < 3; i++) {
+ segment.commit();
+ }
+
+ Assert.assertEquals(lastSize, segment.getCommitPosition());
+ Assert.assertEquals(baseOffset + lastSize, segment.getCommitOffset());
+ Assert.assertEquals(baseOffset + lastSize + MessageBufferUtilTest.MSG_LEN * 3, segment.getMaxOffset());
+
+ segment.blocker.join();
+ segment.blocker = null;
+ segment.commit();
+ Assert.assertEquals(lastSize + MessageBufferUtilTest.MSG_LEN * 2, segment.getCommitPosition());
+ Assert.assertEquals(baseOffset + lastSize + MessageBufferUtilTest.MSG_LEN * 2, segment.getCommitOffset());
Assert.assertEquals(baseOffset + lastSize + MessageBufferUtilTest.MSG_LEN * 3, segment.getMaxOffset());
+
+ segment.commit();
+ Assert.assertEquals(lastSize + MessageBufferUtilTest.MSG_LEN * 3, segment.getCommitPosition());
Assert.assertEquals(baseOffset + lastSize + MessageBufferUtilTest.MSG_LEN * 3, segment.getCommitOffset());
+ Assert.assertEquals(baseOffset + lastSize + MessageBufferUtilTest.MSG_LEN * 3, segment.getMaxOffset());
ByteBuffer msg1 = segment.read(lastSize, MessageBufferUtilTest.MSG_LEN);
Assert.assertEquals(baseOffset + lastSize, MessageBufferUtil.getCommitLogOffset(msg1));
@@ -23,7 +23,7 @@ import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.tieredstore.common.FileSegmentType;
import org.apache.rocketmq.tieredstore.common.TieredMessageStoreConfig;
import org.apache.rocketmq.tieredstore.provider.TieredFileSegment;
-import org.apache.rocketmq.tieredstore.provider.inputstream.TieredFileSegmentInputStream;
+import org.apache.rocketmq.tieredstore.provider.stream.FileSegmentInputStream;
import org.apache.rocketmq.tieredstore.util.TieredStoreUtil;
import org.junit.Assert;
@@ -33,6 +33,8 @@ public class MemoryFileSegment extends TieredFileSegment {
public CompletableFuture<Boolean> blocker;
+ protected int size = 0;
+
protected boolean checkSize = true;
public MemoryFileSegment(FileSegmentType fileType, MessageQueue messageQueue, long baseOffset,
@@ -56,6 +58,18 @@ public class MemoryFileSegment extends TieredFileSegment {
memStore.position((int) getSize());
}
+ public boolean isCheckSize() {
+ return checkSize;
+ }
+
+ public void setCheckSize(boolean checkSize) {
+ this.checkSize = checkSize;
+ }
+
+ public ByteBuffer getMemStore() {
+ return memStore;
+ }
+
@Override
public String getPath() {
return filePath;
@@ -66,7 +80,11 @@ public class MemoryFileSegment extends TieredFileSegment {
if (checkSize) {
return 1000;
}
- return 0;
+ return size;
+ }
+
+ public void setSize(int size) {
+ this.size = size;
}
@Override
@@ -85,11 +103,11 @@ public class MemoryFileSegment extends TieredFileSegment {
@Override
public CompletableFuture<Boolean> commit0(
- TieredFileSegmentInputStream inputStream, long position, int length, boolean append) {
+ FileSegmentInputStream inputStream, long position, int length, boolean append) {
try {
if (blocker != null && !blocker.get()) {
- throw new IllegalStateException();
+ throw new IllegalStateException("Commit Exception for Memory Test");
}
} catch (InterruptedException | ExecutionException e) {
Assert.fail(e.getMessage());
@@ -98,7 +116,6 @@ public class MemoryFileSegment extends TieredFileSegment {
Assert.assertTrue(!checkSize || position >= getSize());
byte[] buffer = new byte[1024];
-
int startPos = memStore.position();
try {
int len;
@@ -22,7 +22,7 @@ import java.util.concurrent.ExecutionException;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.tieredstore.common.FileSegmentType;
import org.apache.rocketmq.tieredstore.common.TieredMessageStoreConfig;
-import org.apache.rocketmq.tieredstore.provider.inputstream.TieredFileSegmentInputStream;
+import org.apache.rocketmq.tieredstore.provider.stream.FileSegmentInputStream;
import org.apache.rocketmq.tieredstore.util.TieredStoreUtil;
import org.junit.Assert;
@@ -46,7 +46,7 @@ public class MemoryFileSegmentWithoutCheck extends MemoryFileSegment {
}
@Override
- public CompletableFuture<Boolean> commit0(TieredFileSegmentInputStream inputStream, long position, int length,
+ public CompletableFuture<Boolean> commit0(FileSegmentInputStream inputStream, long position, int length,
boolean append) {
try {
if (blocker != null && !blocker.get()) {
--
2.32.0.windows.2
From d000ef947d7c99918ceba0fa451c1e29fd84ba07 Mon Sep 17 00:00:00 2001
From: yuz10 <845238369@qq.com>
Date: Thu, 31 Aug 2023 09:41:33 +0800
Subject: [PATCH 3/7] [ISSUE #7283] Incorrect dledger commitlog min offset
after mappedFile re delete failed (#7284)
.../apache/rocketmq/store/dledger/DLedgerCommitLog.java | 7 ++++++-
1 file changed, 6 insertions(+), 1 deletion(-)
@@ -162,7 +162,12 @@ public class DLedgerCommitLog extends CommitLog {
if (!mappedFileQueue.getMappedFiles().isEmpty()) {
return mappedFileQueue.getMinOffset();
}
- return dLedgerFileList.getMinOffset();
+ for (MmapFile file : dLedgerFileList.getMappedFiles()) {
+ if (file.isAvailable()) {
+ return file.getFileFromOffset() + file.getStartPosition();
+ }
+ }
+ return 0;
}
@Override
--
2.32.0.windows.2
From f82718ae3b77a16b553c03f672dc971a2d5d48fa Mon Sep 17 00:00:00 2001
From: cnScarb <jjhfen00@163.com>
Date: Thu, 31 Aug 2023 15:50:10 +0800
Subject: [PATCH 4/7] [ISSUE #7208] fix: when deleting topic also delete its
pop retry topic (#7209)
.../processor/AdminBrokerProcessor.java | 24 ++++++++++---
.../processor/AdminBrokerProcessorTest.java | 36 +++++++++++++++++++
2 files changed, 55 insertions(+), 5 deletions(-)
@@ -51,6 +51,7 @@ import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
import org.apache.rocketmq.broker.transaction.queue.TransactionalMessageUtil;
import org.apache.rocketmq.common.AclConfig;
import org.apache.rocketmq.common.BrokerConfig;
+import org.apache.rocketmq.common.KeyBuilder;
import org.apache.rocketmq.common.LockCallback;
import org.apache.rocketmq.common.MQVersion;
import org.apache.rocketmq.common.MixAll;
@@ -542,16 +543,29 @@ public class AdminBrokerProcessor implements NettyRequestProcessor {
}
}
- this.brokerController.getTopicConfigManager().deleteTopicConfig(requestHeader.getTopic());
- this.brokerController.getTopicQueueMappingManager().delete(requestHeader.getTopic());
- this.brokerController.getConsumerOffsetManager().cleanOffsetByTopic(requestHeader.getTopic());
- this.brokerController.getPopInflightMessageCounter().clearInFlightMessageNumByTopicName(requestHeader.getTopic());
- this.brokerController.getMessageStore().deleteTopics(Sets.newHashSet(requestHeader.getTopic()));
+ final Set<String> groups = this.brokerController.getConsumerOffsetManager().whichGroupByTopic(topic);
+ // delete pop retry topics first
+ for (String group : groups) {
+ final String popRetryTopic = KeyBuilder.buildPopRetryTopic(topic, group);
+ if (brokerController.getTopicConfigManager().selectTopicConfig(popRetryTopic) != null) {
+ deleteTopicInBroker(popRetryTopic);
+ }
+ }
+ // delete topic
+ deleteTopicInBroker(topic);
response.setCode(ResponseCode.SUCCESS);
response.setRemark(null);
return response;
}
+ private void deleteTopicInBroker(String topic) {
+ this.brokerController.getTopicConfigManager().deleteTopicConfig(topic);
+ this.brokerController.getTopicQueueMappingManager().delete(topic);
+ this.brokerController.getConsumerOffsetManager().cleanOffsetByTopic(topic);
+ this.brokerController.getPopInflightMessageCounter().clearInFlightMessageNumByTopicName(topic);
+ this.brokerController.getMessageStore().deleteTopics(Sets.newHashSet(topic));
+ }
+
private synchronized RemotingCommand updateAndCreateAccessConfig(ChannelHandlerContext ctx,
RemotingCommand request) throws RemotingCommandException {
final RemotingCommand response = RemotingCommand.createResponseCommand(null);
@@ -29,6 +29,7 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.LongAdder;
import org.apache.rocketmq.broker.BrokerController;
@@ -41,6 +42,7 @@ import org.apache.rocketmq.broker.topic.RocksDBTopicConfigManager;
import org.apache.rocketmq.broker.topic.TopicConfigManager;
import org.apache.rocketmq.common.BoundaryType;
import org.apache.rocketmq.common.BrokerConfig;
+import org.apache.rocketmq.common.KeyBuilder;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.TopicFilterType;
@@ -90,8 +92,11 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anySet;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@RunWith(MockitoJUnitRunner.class)
@@ -321,6 +326,37 @@ public class AdminBrokerProcessorTest {
"please execute it from master broker.");
}
+ @Test
+ public void testDeleteWithPopRetryTopic() throws Exception {
+ String topic = "topicA";
+ String anotherTopic = "another_topicA";
+
+ topicConfigManager = mock(TopicConfigManager.class);
+ when(brokerController.getTopicConfigManager()).thenReturn(topicConfigManager);
+ final ConcurrentHashMap<String, TopicConfig> topicConfigTable = new ConcurrentHashMap<>();
+ topicConfigTable.put(topic, new TopicConfig());
+ topicConfigTable.put(KeyBuilder.buildPopRetryTopic(topic, "cid1"), new TopicConfig());
+
+ topicConfigTable.put(anotherTopic, new TopicConfig());
+ topicConfigTable.put(KeyBuilder.buildPopRetryTopic(anotherTopic, "cid2"), new TopicConfig());
+ when(topicConfigManager.getTopicConfigTable()).thenReturn(topicConfigTable);
+ when(topicConfigManager.selectTopicConfig(anyString())).thenAnswer(invocation -> {
+ final String selectTopic = invocation.getArgument(0);
+ return topicConfigManager.getTopicConfigTable().get(selectTopic);
+ });
+
+ when(brokerController.getConsumerOffsetManager()).thenReturn(consumerOffsetManager);
+ when(consumerOffsetManager.whichGroupByTopic(topic)).thenReturn(Sets.newHashSet("cid1"));
+
+ RemotingCommand request = buildDeleteTopicRequest(topic);
+ RemotingCommand response = adminBrokerProcessor.processRequest(handlerContext, request);
+ assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
+
+ verify(topicConfigManager).deleteTopicConfig(topic);
+ verify(topicConfigManager).deleteTopicConfig(KeyBuilder.buildPopRetryTopic(topic, "cid1"));
+ verify(messageStore, times(2)).deleteTopics(anySet());
+ }
+
@Test
public void testGetAllTopicConfigInRocksdb() throws Exception {
if (notToBeExecuted()) {
--
2.32.0.windows.2
From 31d10385d1616445478104ce9ef463a8c4852ba2 Mon Sep 17 00:00:00 2001
From: guyinyou <36399867+guyinyou@users.noreply.github.com>
Date: Mon, 4 Sep 2023 14:09:32 +0800
Subject: [PATCH 5/7] [ISSUE #7289] Fixed asynchronous send backpressure
capability
Co-authored-by: guyinyou <guyinyou.gyy@alibaba-inc.com>
.../impl/producer/DefaultMQProducerImpl.java | 77 +++++++++++++------
1 file changed, 53 insertions(+), 24 deletions(-)
@@ -547,6 +547,8 @@ public class DefaultMQProducerImpl implements MQProducerInner {
@Deprecated
public void send(final Message msg, final SendCallback sendCallback, final long timeout)
throws MQClientException, RemotingException, InterruptedException {
+ BackpressureSendCallBack newCallBack = new BackpressureSendCallBack(sendCallback);
+
final long beginStartTime = System.currentTimeMillis();
Runnable runnable = new Runnable() {
@Override
@@ -554,20 +556,53 @@ public class DefaultMQProducerImpl implements MQProducerInner {
long costTime = System.currentTimeMillis() - beginStartTime;
if (timeout > costTime) {
try {
- sendDefaultImpl(msg, CommunicationMode.ASYNC, sendCallback, timeout - costTime);
+ sendDefaultImpl(msg, CommunicationMode.ASYNC, newCallBack, timeout - costTime);
} catch (Exception e) {
- sendCallback.onException(e);
+ newCallBack.onException(e);
}
} else {
- sendCallback.onException(
+ newCallBack.onException(
new RemotingTooMuchRequestException("DEFAULT ASYNC send call timeout"));
}
}
};
- executeAsyncMessageSend(runnable, msg, sendCallback, timeout, beginStartTime);
+ executeAsyncMessageSend(runnable, msg, newCallBack, timeout, beginStartTime);
}
- public void executeAsyncMessageSend(Runnable runnable, final Message msg, final SendCallback sendCallback,
+ class BackpressureSendCallBack implements SendCallback {
+ public boolean isSemaphoreAsyncSizeAquired = false;
+ public boolean isSemaphoreAsyncNumAquired = false;
+ public int msgLen;
+ private final SendCallback sendCallback;
+
+ public BackpressureSendCallBack(final SendCallback sendCallback) {
+ this.sendCallback = sendCallback;
+ }
+
+ @Override
+ public void onSuccess(SendResult sendResult) {
+ if (isSemaphoreAsyncSizeAquired) {
+ semaphoreAsyncSendSize.release(msgLen);
+ }
+ if (isSemaphoreAsyncNumAquired) {
+ semaphoreAsyncSendNum.release();
+ }
+ sendCallback.onSuccess(sendResult);
+ }
+
+ @Override
+ public void onException(Throwable e) {
+ if (isSemaphoreAsyncSizeAquired) {
+ semaphoreAsyncSendSize.release(msgLen);
+ }
+ if (isSemaphoreAsyncNumAquired) {
+ semaphoreAsyncSendNum.release();
+ }
+ sendCallback.onException(e);
+ }
+ }
+
+ public void executeAsyncMessageSend(Runnable runnable, final Message msg, final BackpressureSendCallBack sendCallback,
final long timeout, final long beginStartTime)
throws MQClientException, InterruptedException {
ExecutorService executor = this.getAsyncSenderExecutor();
@@ -595,7 +630,9 @@ public class DefaultMQProducerImpl implements MQProducerInner {
return;
}
}
-
+ sendCallback.isSemaphoreAsyncSizeAquired = isSemaphoreAsyncSizeAquired;
+ sendCallback.isSemaphoreAsyncNumAquired = isSemaphoreAsyncNumAquired;
+ sendCallback.msgLen = msgLen;
executor.submit(runnable);
} catch (RejectedExecutionException e) {
if (isEnableBackpressureForAsyncMode) {
@@ -603,15 +640,7 @@ public class DefaultMQProducerImpl implements MQProducerInner {
} else {
throw new MQClientException("executor rejected ", e);
}
- } finally {
- if (isSemaphoreAsyncSizeAquired) {
- semaphoreAsyncSendSize.release(msgLen);
- }
- if (isSemaphoreAsyncNumAquired) {
- semaphoreAsyncSendNum.release();
- }
}
-
}
public MessageQueue invokeMessageQueueSelector(Message msg, MessageQueueSelector selector, Object arg,
@@ -1188,7 +1217,7 @@ public class DefaultMQProducerImpl implements MQProducerInner {
@Deprecated
public void send(final Message msg, final MessageQueue mq, final SendCallback sendCallback, final long timeout)
throws MQClientException, RemotingException, InterruptedException {
-
+ BackpressureSendCallBack newCallBack = new BackpressureSendCallBack(sendCallback);
final long beginStartTime = System.currentTimeMillis();
Runnable runnable = new Runnable() {
@Override
@@ -1203,22 +1232,22 @@ public class DefaultMQProducerImpl implements MQProducerInner {
long costTime = System.currentTimeMillis() - beginStartTime;
if (timeout > costTime) {
try {
- sendKernelImpl(msg, mq, CommunicationMode.ASYNC, sendCallback, null,
+ sendKernelImpl(msg, mq, CommunicationMode.ASYNC, newCallBack, null,
timeout - costTime);
} catch (MQBrokerException e) {
throw new MQClientException("unknown exception", e);
}
} else {
- sendCallback.onException(new RemotingTooMuchRequestException("call timeout"));
+ newCallBack.onException(new RemotingTooMuchRequestException("call timeout"));
}
} catch (Exception e) {
- sendCallback.onException(e);
+ newCallBack.onException(e);
}
}
};
- executeAsyncMessageSend(runnable, msg, sendCallback, timeout, beginStartTime);
+ executeAsyncMessageSend(runnable, msg, newCallBack, timeout, beginStartTime);
}
/**
@@ -1315,7 +1344,7 @@ public class DefaultMQProducerImpl implements MQProducerInner {
public void send(final Message msg, final MessageQueueSelector selector, final Object arg,
final SendCallback sendCallback, final long timeout)
throws MQClientException, RemotingException, InterruptedException {
-
+ BackpressureSendCallBack newCallBack = new BackpressureSendCallBack(sendCallback);
final long beginStartTime = System.currentTimeMillis();
Runnable runnable = new Runnable() {
@Override
@@ -1324,21 +1353,21 @@ public class DefaultMQProducerImpl implements MQProducerInner {
if (timeout > costTime) {
try {
try {
- sendSelectImpl(msg, selector, arg, CommunicationMode.ASYNC, sendCallback,
+ sendSelectImpl(msg, selector, arg, CommunicationMode.ASYNC, newCallBack,
timeout - costTime);
} catch (MQBrokerException e) {
throw new MQClientException("unknown exception", e);
}
} catch (Exception e) {
- sendCallback.onException(e);
+ newCallBack.onException(e);
}
} else {
- sendCallback.onException(new RemotingTooMuchRequestException("call timeout"));
+ newCallBack.onException(new RemotingTooMuchRequestException("call timeout"));
}
}
};
- executeAsyncMessageSend(runnable, msg, sendCallback, timeout, beginStartTime);
+ executeAsyncMessageSend(runnable, msg, newCallBack, timeout, beginStartTime);
}
/**
--
2.32.0.windows.2
From d67b9d64cbd53824798af57ba18770e0fcefa37a Mon Sep 17 00:00:00 2001
From: yuz10 <845238369@qq.com>
Date: Wed, 6 Sep 2023 14:07:23 +0800
Subject: [PATCH 6/7] [ISSUE #7302] Fix singleTopicRegister code deleted in
merge
.../apache/rocketmq/broker/topic/TopicConfigManager.java | 6 +++++-
1 file changed, 5 insertions(+), 1 deletion(-)
@@ -330,7 +330,11 @@ public class TopicConfigManager extends ConfigManager {
log.error("createTopicIfAbsent ", e);
}
if (createNew && register) {
- this.brokerController.registerIncrementBrokerData(topicConfig, dataVersion);
+ if (brokerController.getBrokerConfig().isEnableSingleTopicRegister()) {
+ this.brokerController.registerSingleTopicAll(topicConfig);
+ } else {
+ this.brokerController.registerIncrementBrokerData(topicConfig, dataVersion);
+ }
}
return getTopicConfig(topicConfig.getTopicName());
}
--
2.32.0.windows.2
From 37017dbaec5c521fd529ef4aecf3658092884f84 Mon Sep 17 00:00:00 2001
From: lizhimins <707364882@qq.com>
Date: Wed, 6 Sep 2023 15:23:15 +0800
Subject: [PATCH 7/7] [ISSUE #7305] Fix metrics and transactional module not
shutdown while broker offline cause coredump(#7307)
.../java/org/apache/rocketmq/broker/BrokerController.java | 8 ++++++++
.../queue/TransactionalMessageServiceImpl.java | 4 +++-
2 files changed, 11 insertions(+), 1 deletion(-)
@@ -1302,6 +1302,10 @@ public class BrokerController {
this.fastRemotingServer.shutdown();
}
+ if (this.brokerMetricsManager != null) {
+ this.brokerMetricsManager.shutdown();
+ }
+
if (this.brokerStatsManager != null) {
this.brokerStatsManager.shutdown();
}
@@ -1324,6 +1328,10 @@ public class BrokerController {
this.ackMessageProcessor.shutdownPopReviveService();
}
+ if (this.transactionalMessageService != null) {
+ this.transactionalMessageService.close();
+ }
+
if (this.notificationProcessor != null) {
this.notificationProcessor.getPopLongPollingService().shutdown();
}
@@ -629,7 +629,9 @@ public class TransactionalMessageServiceImpl implements TransactionalMessageServ
@Override
public void close() {
-
+ if (this.transactionalOpBatchService != null) {
+ this.transactionalOpBatchService.shutdown();
+ }
}
public Message getOpMessage(int queueId, String moreData) {
--
2.32.0.windows.2