mirror of
https://github.com/apache/rocketmq.git
synced 2026-08-28 20:09:14 +08:00
fix: restore divided-commitlog fallback in DLedgerCommitLog#getMaxOffset
getCommittedPos() returns a negative value when no entry has been committed yet, which is exactly the state of a freshly started DLedger store layered on top of legacy commitlog data. The `committedPos == 0` guard therefore skipped the dLedgerFileList.getMinOffset() fallback that was previously applied for any non-positive committed position, so getMaxOffset() and getConfirmOffset() returned 0 instead of the divided commitlog offset. DefaultMessageStore then computed a negative dispatchBehindBytes() and left reputFromOffset behind, failing MixCommitlogTest#testPutAndGet with expected:<0> but was:<-1163232>. Also make StoreTestBase#nextPort() probe that a candidate port can actually be bound before handing it out. The counter previously returned port numbers blindly, so a stale listener in the 30000+ range made callers fail with "Address already in use"; the new three-node DLedger test needs three listeners at once and hit this on CI.
This commit is contained in:
@@ -162,7 +162,11 @@ public class DLedgerCommitLog extends CommitLog {
|
||||
if (committedPos > 0) {
|
||||
return committedPos;
|
||||
}
|
||||
if (committedPos == 0 && dLedgerFileList.getMinOffset() > 0) {
|
||||
// getCommittedPos() yields a non-positive value when no entry has been committed yet,
|
||||
// which is the normal state right after a DLedger store is layered on top of legacy
|
||||
// commitlog data. In that case the divided commitlog offset, i.e. the min offset of the
|
||||
// DLedger file list, is the real max offset.
|
||||
if (dLedgerFileList.getMinOffset() > 0) {
|
||||
return dLedgerFileList.getMinOffset();
|
||||
}
|
||||
return 0;
|
||||
|
||||
@@ -24,8 +24,10 @@ import org.apache.rocketmq.common.message.MessageExtBrokerInner;
|
||||
import org.junit.After;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.net.InetAddress;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.ServerSocket;
|
||||
import java.net.SocketAddress;
|
||||
import java.net.UnknownHostException;
|
||||
import java.util.ArrayList;
|
||||
@@ -47,8 +49,27 @@ public class StoreTestBase {
|
||||
|
||||
private static AtomicInteger port = new AtomicInteger(30000);
|
||||
|
||||
private static final int MAX_PORT_PROBE_ATTEMPTS = 200;
|
||||
|
||||
public static synchronized int nextPort() {
|
||||
return port.addAndGet(5);
|
||||
for (int i = 0; i < MAX_PORT_PROBE_ATTEMPTS; i++) {
|
||||
int candidate = port.addAndGet(5);
|
||||
if (isPortAvailable(candidate)) {
|
||||
return candidate;
|
||||
}
|
||||
}
|
||||
throw new IllegalStateException("Failed to find an available port after "
|
||||
+ MAX_PORT_PROBE_ATTEMPTS + " attempts, last tried " + port.get());
|
||||
}
|
||||
|
||||
private static boolean isPortAvailable(int candidate) {
|
||||
try (ServerSocket serverSocket = new ServerSocket()) {
|
||||
serverSocket.setReuseAddress(false);
|
||||
serverSocket.bind(new InetSocketAddress(candidate));
|
||||
return true;
|
||||
} catch (IOException e) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
protected MessageExtBatch buildBatchMessage(int size) {
|
||||
|
||||
Reference in New Issue
Block a user