This is an automated email from the ASF dual-hosted git repository.
ptupitsyn pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new d809333838 IGNITE-20963 Fix
testClientReceivesPartitionAssignmentUpdates flakiness (#2886)
d809333838 is described below
commit d809333838768cb4c7bb2813a4c51ee1ff3f8a1f
Author: Pavel Tupitsyn <ptupit...@apache.org>
AuthorDate: Wed Nov 29 06:35:54 2023 +0200
IGNITE-20963 Fix testClientReceivesPartitionAssignmentUpdates flakiness
(#2886)
* Use `partitionAssignmentTimestamp` and `waitForCondition` to wait for
heartbeat update reliably
* Fix `FakePlacementDriver` to preserve `leaseStartTime`
* Fix `PartitionAwarenessTest.initPrimaryReplicas` to use the same
`leaseStartTime` for all nodes
---
.../handler/ClientPrimaryReplicaTrackerTest.java | 9 +++---
.../ignite/client/handler/FakePlacementDriver.java | 18 +++++------
.../ignite/internal/client/table/ClientTable.java | 5 ++-
.../ignite/client/PartitionAwarenessTest.java | 36 +++++++++++++++-------
4 files changed, 42 insertions(+), 26 deletions(-)
diff --git
a/modules/client-handler/src/test/java/org/apache/ignite/client/handler/ClientPrimaryReplicaTrackerTest.java
b/modules/client-handler/src/test/java/org/apache/ignite/client/handler/ClientPrimaryReplicaTrackerTest.java
index efd27fd0f9..416bde7435 100644
---
a/modules/client-handler/src/test/java/org/apache/ignite/client/handler/ClientPrimaryReplicaTrackerTest.java
+++
b/modules/client-handler/src/test/java/org/apache/ignite/client/handler/ClientPrimaryReplicaTrackerTest.java
@@ -28,7 +28,6 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicLong;
import
org.apache.ignite.client.handler.ClientPrimaryReplicaTracker.PrimaryReplicasResult;
import org.apache.ignite.internal.TestHybridClock;
-import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.table.IgniteTablesInternal;
import org.apache.ignite.internal.table.InternalTable;
import org.apache.ignite.internal.table.TableViewInternal;
@@ -51,7 +50,7 @@ class ClientPrimaryReplicaTrackerTest extends
BaseIgniteAbstractTest {
@BeforeEach
public void setUp() throws Exception {
driver = new FakePlacementDriver(PARTITIONS);
- driver.setReplicas(List.of("s1", "s2"), TABLE_ID);
+ driver.setReplicas(List.of("s1", "s2"), TABLE_ID, 1);
InternalTable internalTable = mock(InternalTable.class);
when(internalTable.partitions()).thenReturn(PARTITIONS);
@@ -89,7 +88,7 @@ class ClientPrimaryReplicaTrackerTest extends
BaseIgniteAbstractTest {
assertEquals(1, tracker.maxStartTime());
driver.updateReplica("s3", TABLE_ID, 0, 2);
- assertEquals(new HybridTimestamp(2, 0).longValue(),
tracker.maxStartTime());
+ assertEquals(2, tracker.maxStartTime());
PrimaryReplicasResult replicas =
tracker.primaryReplicasAsync(TABLE_ID, null).join();
assertEquals(PARTITIONS, replicas.nodeNames().size());
@@ -105,7 +104,7 @@ class ClientPrimaryReplicaTrackerTest extends
BaseIgniteAbstractTest {
assertEquals(1, tracker.maxStartTime());
driver.updateReplica(null, TABLE_ID, 1, 2);
- assertEquals(new HybridTimestamp(2, 0).longValue(),
tracker.maxStartTime());
+ assertEquals(2, tracker.maxStartTime());
PrimaryReplicasResult replicas =
tracker.primaryReplicasAsync(TABLE_ID, null).join();
assertEquals(PARTITIONS, replicas.nodeNames().size());
@@ -138,7 +137,7 @@ class ClientPrimaryReplicaTrackerTest extends
BaseIgniteAbstractTest {
driver.updateReplica("update-3", TABLE_ID, 0, 15);
driver.updateReplica("old-update-4", TABLE_ID, 0, 14);
- assertEquals(new HybridTimestamp(15, 0).longValue(),
tracker.maxStartTime());
+ assertEquals(15, tracker.maxStartTime());
PrimaryReplicasResult replicas =
tracker.primaryReplicasAsync(TABLE_ID, null).join();
assertEquals(PARTITIONS, replicas.nodeNames().size());
diff --git
a/modules/client-handler/src/testFixtures/java/org/apache/ignite/client/handler/FakePlacementDriver.java
b/modules/client-handler/src/testFixtures/java/org/apache/ignite/client/handler/FakePlacementDriver.java
index 2a349d10d0..a43b052ee0 100644
---
a/modules/client-handler/src/testFixtures/java/org/apache/ignite/client/handler/FakePlacementDriver.java
+++
b/modules/client-handler/src/testFixtures/java/org/apache/ignite/client/handler/FakePlacementDriver.java
@@ -38,13 +38,13 @@ public class FakePlacementDriver extends
AbstractEventProducer<PrimaryReplicaEve
implements PlacementDriver {
private final int partitions;
- private final List<String> primaryReplicas;
+ private final List<ReplicaMeta> primaryReplicas;
private boolean returnError;
public FakePlacementDriver(int partitions) {
this.partitions = partitions;
- primaryReplicas = new ArrayList<>(Collections.nCopies(partitions,
"s"));
+ primaryReplicas = new ArrayList<>(Collections.nCopies(partitions,
getReplicaMeta("s", HybridTimestamp.MIN_VALUE.longValue())));
}
public void returnError(boolean returnError) {
@@ -54,12 +54,12 @@ public class FakePlacementDriver extends
AbstractEventProducer<PrimaryReplicaEve
/**
* Sets all primary replicas.
*/
- public void setReplicas(List<String> replicas, int tableId) {
+ public void setReplicas(List<String> replicas, int tableId, long
leaseStartTime) {
assert replicas.size() == partitions;
for (int partition = 0; partition < replicas.size(); partition++) {
String replica = replicas.get(partition);
- updateReplica(replica, tableId, partition,
System.currentTimeMillis());
+ updateReplica(replica, tableId, partition, leaseStartTime);
}
}
@@ -67,11 +67,11 @@ public class FakePlacementDriver extends
AbstractEventProducer<PrimaryReplicaEve
* Sets primary replica for the given partition.
*/
public void updateReplica(String replica, int tableId, int partition, long
leaseStartTime) {
- primaryReplicas.set(partition, replica);
+ primaryReplicas.set(partition, getReplicaMeta(replica,
leaseStartTime));
TablePartitionId groupId = new TablePartitionId(tableId, partition);
PrimaryReplicaEventParameters params = new
PrimaryReplicaEventParameters(
- 0, groupId, replica, new HybridTimestamp(leaseStartTime, 0));
+ 0, groupId, replica,
HybridTimestamp.hybridTimestamp(leaseStartTime));
fireEvent(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED, params);
}
@@ -83,7 +83,7 @@ public class FakePlacementDriver extends
AbstractEventProducer<PrimaryReplicaEve
return returnError
? CompletableFuture.failedFuture(new
RuntimeException("FakePlacementDriver expected error"))
- :
CompletableFuture.completedFuture(getReplicaMeta(primaryReplicas.get(id.partitionId())));
+ :
CompletableFuture.completedFuture(primaryReplicas.get(id.partitionId()));
}
@Override
@@ -96,7 +96,7 @@ public class FakePlacementDriver extends
AbstractEventProducer<PrimaryReplicaEve
return CompletableFuture.completedFuture(null);
}
- private static ReplicaMeta getReplicaMeta(String leaseholder) {
+ private static ReplicaMeta getReplicaMeta(String leaseholder, long
leaseStartTime) {
//noinspection serial
return new ReplicaMeta() {
@Override
@@ -111,7 +111,7 @@ public class FakePlacementDriver extends
AbstractEventProducer<PrimaryReplicaEve
@Override
public HybridTimestamp getStartTime() {
- return HybridTimestamp.MIN_VALUE;
+ return HybridTimestamp.hybridTimestamp(leaseStartTime);
}
@Override
diff --git
a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java
b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java
index 1ee0508999..51ffb9e35a 100644
---
a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java
+++
b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java
@@ -551,7 +551,10 @@ public class ClientTable implements Table {
}
// Returned timestamp can be newer than requested.
- partitionAssignment.timestamp = r.in().unpackLong();
+ long ts = r.in().unpackLong();
+ assert ts >= timestamp : "Returned timestamp is older
than requested: " + ts + " < " + timestamp;
+
+ partitionAssignment.timestamp = ts;
List<String> res = new ArrayList<>(cnt);
for (int i = 0; i < cnt; i++) {
diff --git
a/modules/client/src/test/java/org/apache/ignite/client/PartitionAwarenessTest.java
b/modules/client/src/test/java/org/apache/ignite/client/PartitionAwarenessTest.java
index 052913d118..b0af679313 100644
---
a/modules/client/src/test/java/org/apache/ignite/client/PartitionAwarenessTest.java
+++
b/modules/client/src/test/java/org/apache/ignite/client/PartitionAwarenessTest.java
@@ -17,6 +17,8 @@
package org.apache.ignite.client;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.greaterThan;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -35,7 +37,9 @@ import org.apache.ignite.client.fakes.FakeIgniteTables;
import org.apache.ignite.client.fakes.FakeInternalTable;
import org.apache.ignite.client.handler.FakePlacementDriver;
import org.apache.ignite.compute.IgniteCompute;
+import org.apache.ignite.internal.client.ReliableChannel;
import org.apache.ignite.internal.client.tx.ClientTransaction;
+import org.apache.ignite.internal.hlc.HybridClockImpl;
import org.apache.ignite.internal.table.TableViewInternal;
import org.apache.ignite.internal.testframework.IgniteTestUtils;
import org.apache.ignite.internal.util.IgniteUtils;
@@ -71,9 +75,9 @@ public class PartitionAwarenessTest extends
AbstractClientTest {
private static IgniteClient client2;
- private @Nullable String lastOp;
+ private volatile @Nullable String lastOp;
- private @Nullable String lastOpServerName;
+ private volatile @Nullable String lastOpServerName;
private static final AtomicInteger nextTableId = new AtomicInteger(101);
@@ -166,6 +170,8 @@ public class PartitionAwarenessTest extends
AbstractClientTest {
@ParameterizedTest
@ValueSource(booleans = {true, false})
public void testClientReceivesPartitionAssignmentUpdates(boolean
useHeartbeat) throws InterruptedException {
+ ReliableChannel ch = IgniteTestUtils.getFieldValue(client2, "ch");
+
// Check default assignment.
RecordView<Tuple> recordView = defaultTable().recordView();
@@ -173,19 +179,25 @@ public class PartitionAwarenessTest extends
AbstractClientTest {
assertOpOnNode(nodeKey2, "get", x -> recordView.get(null,
Tuple.create().set("ID", 2L)));
// Update partition assignment.
+ var oldTs = ch.partitionAssignmentTimestamp();
initPrimaryReplicas(reversedReplicas());
if (useHeartbeat) {
// Wait for heartbeat message to receive change notification flag.
- Thread.sleep(500);
+ assertTrue(IgniteTestUtils.waitForCondition(() ->
ch.partitionAssignmentTimestamp() > oldTs, 3000));
} else {
- // Perform a request on the default channel to receive change
notification flag.
- // Use two requests because of round-robin.
- client2.tables().tables();
- client2.tables().tables();
+ // Perform requests to receive change notification flag.
+ int maxRequests = 50;
+ while (ch.partitionAssignmentTimestamp() <= oldTs && maxRequests--
> 0) {
+ client2.tables().tables();
+ }
+
+ assertThat("Failed to receive assignment update", maxRequests,
greaterThan(0));
}
// Check new assignment.
+ assertThat(ch.partitionAssignmentTimestamp(), greaterThan(oldTs));
+
assertOpOnNode(nodeKey2, "get", x -> recordView.get(null,
Tuple.create().set("ID", 1L)));
assertOpOnNode(nodeKey1, "get", x -> recordView.get(null,
Tuple.create().set("ID", 2L)));
}
@@ -611,16 +623,18 @@ public class PartitionAwarenessTest extends
AbstractClientTest {
}
private static void initPrimaryReplicas(@Nullable List<String> replicas) {
- initPrimaryReplicas(testServer.placementDriver(), replicas);
- initPrimaryReplicas(testServer2.placementDriver(), replicas);
+ long leaseStartTime = new HybridClockImpl().nowLong();
+
+ initPrimaryReplicas(testServer.placementDriver(), replicas,
leaseStartTime);
+ initPrimaryReplicas(testServer2.placementDriver(), replicas,
leaseStartTime);
}
- private static void initPrimaryReplicas(FakePlacementDriver
placementDriver, @Nullable List<String> replicas) {
+ private static void initPrimaryReplicas(FakePlacementDriver
placementDriver, @Nullable List<String> replicas, long leaseStartTime) {
if (replicas == null) {
replicas = defaultReplicas();
}
- placementDriver.setReplicas(replicas, nextTableId.get() - 1);
+ placementDriver.setReplicas(replicas, nextTableId.get() - 1,
leaseStartTime);
}
private static List<String> defaultReplicas() {
