From e8c0d1b865426420b4bd9c2e482b27be0e4df57f Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Sat, 11 Jul 2026 10:37:27 +0000
Subject: [PATCH] [#710] Fix replication catch-up re-sending updates with the original assured flag (#714)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/server/SafeDataExpectedAcksInfo.java | 15 ++
opendj-server-legacy/src/test/java/org/opends/server/replication/server/AssuredReplicationServerTest.java | 121 ++++++++++++++++
opendj-server-legacy/src/main/java/org/opends/server/replication/server/ExpectedAcksInfo.java | 29 ++++
opendj-server-legacy/src/test/java/org/opends/server/replication/server/ExpectedAcksInfoTest.java | 85 ++++++++++++
opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/AckMsg.java | 7 +
opendj-server-legacy/src/main/java/org/opends/server/replication/server/SafeReadExpectedAcksInfo.java | 15 ++
opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerHandler.java | 56 +++++++
opendj-server-legacy/src/main/java/org/opends/server/replication/server/MessageHandler.java | 27 +++
opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java | 16 ++
9 files changed, 362 insertions(+), 9 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/AckMsg.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/AckMsg.java
index 357d9cb..1c96fdf 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/AckMsg.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/AckMsg.java
@@ -13,6 +13,7 @@
*
* Copyright 2006-2009 Sun Microsystems, Inc.
* Portions Copyright 2013-2015 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC
*/
package org.opends.server.replication.protocol;
@@ -254,5 +255,11 @@
"concerned server ids: " + idList;
}
+ /** {@inheritDoc} */
+ @Override
+ public String toString()
+ {
+ return getClass().getSimpleName() + " csn: " + csn + ", " + errorsToString();
+ }
}
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ExpectedAcksInfo.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ExpectedAcksInfo.java
index 3136fb1..05aad14 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ExpectedAcksInfo.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ExpectedAcksInfo.java
@@ -13,13 +13,17 @@
*
* Copyright 2008-2009 Sun Microsystems, Inc.
* Portions Copyright 2013-2015 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC
*/
package org.opends.server.replication.server;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import org.opends.server.replication.common.AssuredMode;
import org.opends.server.replication.common.CSN;
@@ -69,6 +73,15 @@
protected Map<Integer,Boolean> expectedServersAckStatus = new HashMap<>();
/**
+ * Immutable snapshot of the ids of the servers we expect an ack from, taken
+ * at construction time. Used for lock-free membership checks: the key set of
+ * {@link #expectedServersAckStatus} never changes after construction, but
+ * that map has its values mutated under lock by {@code processReceivedAck()},
+ * so it must not be read concurrently without synchronization.
+ */
+ private final Set<Integer> expectedServerIds;
+
+ /**
* Facility for monitoring:
* If the timeout occurs for the original update, we call createAck(true)
* in the timeout code for sending back an error ack to the original server.
@@ -99,6 +112,22 @@
{
expectedServersAckStatus.put(serverId, false);
}
+ this.expectedServerIds =
+ Collections.unmodifiableSet(new HashSet<>(expectedServers));
+ }
+
+ /**
+ * Indicates whether the provided server is one of the servers an ack is
+ * expected from for the matching update message. Reads an immutable snapshot
+ * taken at construction time, so it is safe to call without holding the lock
+ * on this object.
+ *
+ * @param serverId The serverId of the server.
+ * @return true if an ack is expected from the provided server.
+ */
+ public boolean isExpectedServer(int serverId)
+ {
+ return expectedServerIds.contains(serverId);
}
/**
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/MessageHandler.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/MessageHandler.java
index 6e280e8..b6e1b2a 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/MessageHandler.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/MessageHandler.java
@@ -80,6 +80,13 @@
private final int maxQueueBytesSize;
/** Specifies whether the consumer is following the producer (is not late). */
private boolean following;
+ /**
+ * Specifies whether the last update message returned by
+ * {@link #getNextMessage()} was re-read from the changelog DB (catch-up
+ * path) rather than taken from the in-memory {@link #msgQueue}. Only ever
+ * accessed by the single consumer thread calling {@link #getNextMessage()}.
+ */
+ private boolean lastMessageFromLateQueue;
/** Specifies the current serverState of this handler. */
private ServerState serverState;
/** Specifies the baseDN of the domain. */
@@ -226,6 +233,21 @@
}
/**
+ * Indicates whether the last update message returned by
+ * {@link #getNextMessage()} was re-read from the changelog DB (catch-up
+ * path) rather than taken from the in-memory queue.
+ * <p>
+ * Must only be called from the consumer thread calling
+ * {@link #getNextMessage()}.
+ *
+ * @return true if the last returned update message came from the late queue
+ */
+ protected boolean isLastMessageFromLateQueue()
+ {
+ return lastMessageFromLateQueue;
+ }
+
+ /**
* Retrieves the name of this monitor provider. It should be unique among all
* monitor providers, including all instances of the same monitor provider.
*
@@ -319,6 +341,9 @@
msgQueue.consumeUpTo(msg);
if (updateServerState(msg))
{
+ // the returned instance is the one re-read from the
+ // changelog DB, not its msgQueue equivalent
+ lastMessageFromLateQueue = true;
return msg;
}
}
@@ -347,6 +372,7 @@
}
if (updateServerState(msg))
{
+ lastMessageFromLateQueue = true;
return msg;
}
continue;
@@ -379,6 +405,7 @@
* by the other server.
* Otherwise just loop to select the next message.
*/
+ lastMessageFromLateQueue = false;
return msg;
}
}
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java
index ebe7346..81d6e48 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ReplicationServerDomain.java
@@ -13,7 +13,7 @@
*
* Copyright 2006-2010 Sun Microsystems, Inc.
* Portions Copyright 2011-2016 ForgeRock AS.
- * Portions Copyrighted 2026 3A Systems, LLC.
+ * Portions Copyright 2026 3A Systems, LLC
*/
package org.opends.server.replication.server;
@@ -790,6 +790,20 @@
}
/**
+ * Indicates whether the ack window for the provided CSN is still open and
+ * the provided server is one of the servers an ack is expected from.
+ *
+ * @param csn The CSN of the update message.
+ * @param serverId The serverId of the candidate acknowledging server.
+ * @return true if an ack from the provided server would be accounted for.
+ */
+ boolean isExpectedAck(CSN csn, int serverId)
+ {
+ final ExpectedAcksInfo expectedAcksInfo = waitingAcks.get(csn);
+ return expectedAcksInfo != null && expectedAcksInfo.isExpectedServer(serverId);
+ }
+
+ /**
* Process an ack received from a given server.
*
* @param ack The ack message received.
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/SafeDataExpectedAcksInfo.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/SafeDataExpectedAcksInfo.java
index 38e8a7e..4c5d179 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/SafeDataExpectedAcksInfo.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/SafeDataExpectedAcksInfo.java
@@ -13,6 +13,7 @@
*
* Copyright 2008-2009 Sun Microsystems, Inc.
* Portions Copyright 2013-2015 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC
*/
package org.opends.server.replication.server;
@@ -84,7 +85,19 @@
// Get the ack status for the matching server
int ackingServerId = ackingServer.getServerId();
- boolean ackReceived = expectedServersAckStatus.get(ackingServerId);
+ Boolean ackReceived = expectedServersAckStatus.get(ackingServerId);
+ if (ackReceived == null)
+ {
+ // Ack from a server we were not expecting an ack from, for instance
+ // because the update was delivered to it with the assured flag of the
+ // original sender through the changelog catch-up path: ignore it.
+ if (logger.isTraceEnabled())
+ {
+ logger.trace("Received ack from not expected server id: " +
+ ackingServerId + " ack message: " + ackMsg);
+ }
+ return false;
+ }
if (ackReceived)
{
// Sanity check: this should never happen
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/SafeReadExpectedAcksInfo.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/SafeReadExpectedAcksInfo.java
index ae886fa..6a0fbb7 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/SafeReadExpectedAcksInfo.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/SafeReadExpectedAcksInfo.java
@@ -13,6 +13,7 @@
*
* Copyright 2008-2009 Sun Microsystems, Inc.
* Portions Copyright 2013-2015 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC
*/
package org.opends.server.replication.server;
@@ -146,7 +147,19 @@
{
// Get the ack status for the matching server
int ackingServerId = ackingServer.getServerId();
- boolean ackReceived = expectedServersAckStatus.get(ackingServerId);
+ Boolean ackReceived = expectedServersAckStatus.get(ackingServerId);
+ if (ackReceived == null)
+ {
+ // Ack from a server we were not expecting an ack from, for instance
+ // because the update was delivered to it with the assured flag of the
+ // original sender through the changelog catch-up path: ignore it.
+ if (logger.isTraceEnabled())
+ {
+ logger.trace("Received ack from not expected server id: "
+ + ackingServerId + " ack message: " + ackMsg);
+ }
+ return false;
+ }
if (ackReceived)
{
// Sanity check: this should never happen
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerHandler.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerHandler.java
index ca65e02..7a5cb4c 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerHandler.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerHandler.java
@@ -13,12 +13,15 @@
*
* Copyright 2006-2010 Sun Microsystems, Inc.
* Portions Copyright 2011-2016 ForgeRock AS.
+ * Portions Copyright 2026 3A Systems, LLC
*/
package org.opends.server.replication.server;
import static org.opends.messages.ReplicationMessages.*;
+import static org.opends.server.util.StaticUtils.*;
import java.io.IOException;
+import java.io.UnsupportedEncodingException;
import java.util.Random;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
@@ -924,12 +927,32 @@
*/
public UpdateMsg take() throws ChangelogException
{
- final UpdateMsg msg = getNextMessage();
+ UpdateMsg msg = getNextMessage();
+ final boolean fromLateQueue = isLastMessageFromLateQueue();
acquirePermitInSendWindow();
if (msg != null)
{
+ // Updates re-read from the changelog DB (catch-up path) carry the
+ // assured flag, mode and safe data level of their original sender:
+ // the NotAssuredUpdateMsg substitution performed at publish time by
+ // ReplicationServerDomain.addUpdate() only exists on the in-memory
+ // queue path. Normalize them here: keep the assured flag only while
+ // the ack window is still open and this server is expected to ack.
+ // Messages taken from the in-memory queue already carry the
+ // publish-time decision and must NOT be revisited: the ack window may
+ // legitimately close (timeout, or enough acks already received)
+ // before a slow peer gets here - e.g. when acquirePermitInSendWindow()
+ // above blocks on a closed send window - and such a peer must still
+ // receive the assured flag it was deemed eligible for. Its late ack is
+ // then safely ignored by ReplicationServerDomain.processAck() and
+ // ExpectedAcksInfo.processReceivedAck().
+ if (fromLateQueue && msg.isAssured()
+ && !replicationServerDomain.isExpectedAck(msg.getCSN(), serverId))
+ {
+ msg = toNotAssuredUpdateMsg(msg);
+ }
incrementOutCount();
if (msg.isAssured())
{
@@ -940,6 +963,37 @@
return null;
}
+ /**
+ * Substitutes a not assured version of the provided update message so that a
+ * peer not expected to acknowledge it does not receive it with the assured
+ * flag.
+ * <p>
+ * This is the counterpart, for the changelog catch-up path, of the
+ * {@link NotAssuredUpdateMsg} substitution performed on the in-memory queue
+ * path by ReplicationServerDomain.addUpdate(): updates re-read from the
+ * changelog DB keep the assured flag of their original sender and must be
+ * normalized in {@link #take()} before being handed to the ServerWriter.
+ */
+ private UpdateMsg toNotAssuredUpdateMsg(UpdateMsg msg)
+ {
+ try
+ {
+ return new NotAssuredUpdateMsg(msg);
+ }
+ catch (UnsupportedEncodingException e)
+ {
+ // Could not build the not assured form (unexpected message encoding).
+ // Deliver the original message rather than dropping it: losing the
+ // update would break replication consistency, which is worse than a
+ // spurious ack - and such an ack is now safely ignored by
+ // ExpectedAcksInfo.processReceivedAck().
+ logger.error(LocalizableMessage.raw(
+ "Could not substitute a not assured version of update message %s: %s",
+ msg, stackTraceToSingleLineString(e)));
+ return msg;
+ }
+ }
+
private void acquirePermitInSendWindow()
{
boolean acquired = false;
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/AssuredReplicationServerTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/AssuredReplicationServerTest.java
index dd602dc..b58b9cd 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/AssuredReplicationServerTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/AssuredReplicationServerTest.java
@@ -13,7 +13,7 @@
*
* Copyright 2008-2010 Sun Microsystems, Inc.
* Portions Copyright 2011-2016 ForgeRock AS.
- * Portions Copyright 2026 3A Systems, LLC
+ * Portions Copyright 2023-2026 3A Systems, LLC
*/
package org.opends.server.replication.server;
@@ -51,6 +51,7 @@
import org.opends.server.TestCaseUtils;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.AssuredMode;
+import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.CSNGenerator;
import org.opends.server.replication.common.DSInfo;
import org.opends.server.replication.common.RSInfo;
@@ -581,7 +582,7 @@
* Sends a new update from this DS.
* @throws TimeoutException If timeout waiting for an assured ack
*/
- private void sendNewFakeUpdate() throws TimeoutException
+ private CSN sendNewFakeUpdate() throws TimeoutException
{
// Create a new delete update message (the simplest to create)
DeleteMsg delMsg = new DeleteMsg(getBaseDN(), gen.newCSN(), UUID.randomUUID().toString());
@@ -590,6 +591,7 @@
prepareWaitForAckIfAssuredEnabled(delMsg);
publish(delMsg);
waitForAckIfAssuredEnabled(delMsg);
+ return delMsg.getCSN();
}
private void assertReceivedWrongUpdates(int expectedNbUpdates, int expectedNbWrongUpdates)
@@ -751,9 +753,14 @@
private CSNGenerator gen;
/** False if a received update had assured parameters not as expected. */
- private boolean everyUpdatesAreOk = true;
- /** Number of received updates. */
- private int nReceivedUpdates;
+ private volatile boolean everyUpdatesAreOk = true;
+ /**
+ * Number of received updates. Volatile as it is incremented by the listen
+ * thread and polled from the test thread (waitForReceivedUpdates()); a read
+ * observing the incremented value also publishes everyUpdatesAreOk, which
+ * is written before the increment.
+ */
+ private volatile int nReceivedUpdates;
/**
* True if an ack has been replied to a received assured update (in assured
@@ -1076,6 +1083,110 @@
}
/**
+ * Parameters for {@link #testCatchUpClearsAssuredFlag}: assured mode and
+ * safe data level of the update a peer catches up from the changelog.
+ */
+ @DataProvider(name = "catchUpAssuredModeProvider")
+ private Object[][] catchUpAssuredModeProvider()
+ {
+ return new Object[][]
+ {
+ { AssuredMode.SAFE_DATA_MODE, 1 },
+ { AssuredMode.SAFE_DATA_MODE, 2 },
+ { AssuredMode.SAFE_READ_MODE, 1 },
+ };
+ }
+
+ /**
+ * Regression test for issue #710: an update served to a peer through the
+ * changelog catch-up path must not keep the assured flag of its original
+ * sender.
+ * <p>
+ * A fake RS connects with an empty server state after an assured update has
+ * already been received and persisted by the real RS. As the fake RS was not
+ * connected when the update was received, the update is re-read from the
+ * changelog DB (catch-up path) rather than served from the in-memory queue,
+ * and the fake RS is not an expected ack server for it. The update must
+ * therefore be forwarded with its assured flag cleared (while its assured
+ * mode and safe data level are preserved). This used to be delivered with
+ * assured=true, making the fake RS reply a spurious ack. Exercised across
+ * safe data (level 1 and level > 1) and safe read modes.
+ */
+ @Test(dataProvider = "catchUpAssuredModeProvider", enabled = true)
+ public void testCatchUpClearsAssuredFlag(AssuredMode assuredMode,
+ int safeDataLevel) throws Exception
+ {
+ String testCase = "testCatchUpClearsAssuredFlag";
+ debugInfo("Starting " + testCase + " mode=" + assuredMode + " sdl=" + safeDataLevel);
+ initTest();
+ try
+ {
+ // Real RS to be tested
+ rs1 = createReplicationServer(RS1_ID, DEFAULT_GID, SMALL_TIMEOUT, testCase, 0);
+
+ // Main DS sends an assured update, acknowledged by the RS. No other peer
+ // is connected yet, so the update only lands in the changelog DB, not in
+ // any peer message queue.
+ fakeRDs[1] = createFakeReplicationDomain(FDS1_ID, DEFAULT_GID, RS1_ID,
+ DEFAULT_GENID, assuredMode, safeDataLevel, LONG_TIMEOUT, TIMEOUT_DS_SCENARIO);
+ CSN csn = fakeRDs[1].sendNewFakeUpdate();
+
+ // Ensure the update is persisted before the peer connects, so the peer is
+ // served through the changelog catch-up path (the RS acks the DS before
+ // persisting, so returning from sendNewFakeUpdate() does not guarantee it).
+ waitForChangePersisted(rs1, csn);
+
+ // A fake RS connects with an empty state: it must catch up the historical
+ // update from the changelog DB. It expects a non-assured forward and
+ // never replies an ack (TIMEOUT_RS_SCENARIO).
+ fakeRs1 = createFakeReplicationServer(FRS1_ID, DEFAULT_GID, DEFAULT_GENID,
+ false, assuredMode, safeDataLevel, TIMEOUT_RS_SCENARIO);
+
+ // The historical update must have been received, with assured flag off
+ waitForReceivedUpdates(fakeRs1, 1);
+ fakeRs1.assertReceivedUpdates(1);
+ }
+ finally
+ {
+ endTest();
+ }
+ }
+
+ /**
+ * Waits until the provided change has been persisted in the changelog of the
+ * provided replication server, i.e. is retrievable through the catch-up path.
+ */
+ private void waitForChangePersisted(ReplicationServer rs, CSN csn) throws Exception
+ {
+ ReplicationServerDomain domain =
+ rs.getReplicationServerDomain(DN.valueOf(TEST_ROOT_DN_STRING));
+ assertNotNull(domain);
+ int i = 0;
+ while (!domain.getLatestServerState().cover(csn))
+ {
+ if (i++ > 50)
+ {
+ Assert.fail("Change " + csn + " was not persisted in the changelog in time.");
+ }
+ Thread.sleep(100);
+ }
+ }
+
+ /**
+ * Waits until the provided fake RS has received at least the expected number
+ * of updates, so the assertion on the updates does not race with delivery.
+ */
+ private void waitForReceivedUpdates(FakeReplicationServer fakeRs, int expected)
+ throws Exception
+ {
+ int i = 0;
+ while (fakeRs.nReceivedUpdates < expected && i++ <= 50)
+ {
+ Thread.sleep(100);
+ }
+ }
+
+ /**
* Returns possible combinations of parameters for testSafeDataLevelOne test.
*/
@DataProvider(name = "testSafeDataLevelOneProvider")
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ExpectedAcksInfoTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ExpectedAcksInfoTest.java
new file mode 100644
index 0000000..60adb65
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/server/ExpectedAcksInfoTest.java
@@ -0,0 +1,85 @@
+/*
+ * The contents of this file are subject to the terms of the Common Development and
+ * Distribution License (the License). You may not use this file except in compliance with the
+ * License.
+ *
+ * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
+ * specific language governing permission and limitations under the License.
+ *
+ * When distributing Covered Software, include this CDDL Header Notice in each file and include
+ * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
+ * Header, with the fields enclosed by brackets [] replaced by your own identifying
+ * information: "Portions Copyright [year] [name of copyright owner]".
+ *
+ * Portions Copyright 2026 3A Systems, LLC
+ */
+package org.opends.server.replication.server;
+
+import static org.mockito.Mockito.*;
+import static org.testng.Assert.*;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.opends.server.DirectoryServerTestCase;
+import org.opends.server.replication.common.CSN;
+import org.opends.server.replication.protocol.AckMsg;
+import org.testng.annotations.Test;
+
+/**
+ * Test the handling of acks received from servers that were not expected to
+ * acknowledge an assured update.
+ * <p>
+ * A peer served through the changelog catch-up path receives the update with
+ * the assured flag of the original sender (see issue #710), so it may send
+ * back an ack for a server id that is not part of the expected servers of the
+ * matching {@link ExpectedAcksInfo}. That ack must be ignored instead of
+ * raising a {@link NullPointerException} while auto-unboxing the missing map
+ * entry, which would close the connection to the acking peer.
+ */
+@SuppressWarnings("javadoc")
+public class ExpectedAcksInfoTest extends DirectoryServerTestCase
+{
+ private static final CSN CSN1 = new CSN(1, 1, 10);
+
+ private static ServerHandler mockServerHandler(int serverId, boolean isDataServer)
+ {
+ ServerHandler handler = mock(ServerHandler.class);
+ when(handler.getServerId()).thenReturn(serverId);
+ when(handler.isDataServer()).thenReturn(isDataServer);
+ return handler;
+ }
+
+ @Test
+ public void safeReadAckFromNotExpectedServerIsIgnored()
+ {
+ ServerHandler requester = mockServerHandler(1, true);
+ List<Integer> expectedServers = Arrays.asList(2, 3);
+ SafeReadExpectedAcksInfo acksInfo = new SafeReadExpectedAcksInfo(
+ CSN1, requester, expectedServers, Collections.<Integer> emptyList());
+
+ // Server id 99 is not in the expected servers list
+ ServerHandler notExpected = mockServerHandler(99, false);
+ assertFalse(acksInfo.processReceivedAck(notExpected, new AckMsg(CSN1)),
+ "An ack from a not expected server must not complete the ack info");
+ assertFalse(acksInfo.isExpectedServer(99));
+ assertTrue(acksInfo.isExpectedServer(2));
+ }
+
+ @Test
+ public void safeDataAckFromNotExpectedServerIsIgnored()
+ {
+ ServerHandler requester = mockServerHandler(1, true);
+ List<Integer> expectedServers = Arrays.asList(2, 3);
+ SafeDataExpectedAcksInfo acksInfo = new SafeDataExpectedAcksInfo(
+ CSN1, requester, (byte) 3, expectedServers);
+
+ // Server id 99 is a RS not in the expected servers list
+ ServerHandler notExpected = mockServerHandler(99, false);
+ assertFalse(acksInfo.processReceivedAck(notExpected, new AckMsg(CSN1)),
+ "An ack from a not expected server must not complete the ack info");
+ assertFalse(acksInfo.isExpectedServer(99));
+ assertTrue(acksInfo.isExpectedServer(3));
+ }
+}
--
Gitblit v1.10.0