You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "JAkutenshi (via GitHub)" <gi...@apache.org> on 2025/06/09 07:56:07 UTC
[PR] IGNITE-24770 Port tests related to partition lifecycle to PartitionReplicaLifecycleManager tests [ignite-3]
JAkutenshi opened a new pull request, #6004:
URL: https://github.com/apache/ignite-3/pull/6004
JIRA Ticket: [IGNITE-24770](https://issues.apache.org/jira/browse/IGNITE-24770)
## The goal
## The reason
## The solution
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-24770 Copy PartitionReplicaListenerTest logic to test ZonePartitionReplicaListener [ignite-3]
Posted by "JAkutenshi (via GitHub)" <gi...@apache.org>.
JAkutenshi commented on code in PR #6004:
URL: https://github.com/apache/ignite-3/pull/6004#discussion_r2137573025
##########
modules/partition-replicator/src/test/java/org/apache/ignite/internal/partition/replicator/raft/ZonePartitionRaftListenerTest.java:
##########
@@ -112,6 +145,11 @@ class ZonePartitionRaftListenerTest extends BaseIgniteAbstractTest {
@InjectExecutorService
private ExecutorService executor;
+ @Mock
+ private SafeTimeValuesTracker safeTimeClock;
Review Comment:
Fixed.
##########
modules/partition-replicator/src/test/java/org/apache/ignite/internal/partition/replicator/raft/ZonePartitionRaftListenerTest.java:
##########
@@ -367,6 +405,151 @@ void processorsAreNotInitializedWithoutSnapshot(@Mock RaftTableProcessor tablePr
verify(tableProcessor, never()).initialize(any(), any(), anyLong(), anyLong());
}
+ @Test
+ void testSkipWriteCommandByAppliedIndex() {
+ mvPartitionStorage = spy(new TestMvPartitionStorage(PARTITION_ID));
+
+ PartitionListener tableProcessor = partitionListener(TABLE_ID);
+
+ listener.addTableProcessor(TABLE_ID, tableProcessor);
+ // Update(All)Command handling requires both information about raft group topology and the primary replica,
+ // thus onConfigurationCommited and primaryReplicaChangeCommand are called.
+ AtomicInteger raftIndex = new AtomicInteger();
+
+ long index = raftIndex.incrementAndGet();
+
+ listener.onConfigurationCommitted(
+ new RaftGroupConfiguration(
+ index,
+ 1,
+ List.of("foo"),
+ List.of("bar"),
+ null,
+ null
+ ),
+ index,
+ 1
+ );
+
+ PrimaryReplicaChangeCommand command = REPLICA_MESSAGES_FACTORY.primaryReplicaChangeCommand()
+ .primaryReplicaNodeName("primary")
+ .primaryReplicaNodeId(randomUUID())
+ .leaseStartTime(HybridTimestamp.MIN_VALUE.addPhysicalTime(1).longValue())
+ .build();
+
+ listener.onWrite(List.of(writeCommandCommandClosure(raftIndex.incrementAndGet(), 1, command, null, null)).iterator());
+
+ mvPartitionStorage.lastApplied(10L, 1L);
+
+ UpdateCommand updateCommand = mock(UpdateCommand.class);
+ when(updateCommand.tableId()).thenReturn(TABLE_ID);
+
+ WriteIntentSwitchCommand writeIntentSwitchCommand = mock(WriteIntentSwitchCommand.class);
+
+ SafeTimeSyncCommand safeTimeSyncCommand = mock(SafeTimeSyncCommand.class);
+
+ FinishTxCommand finishTxCommand = mock(FinishTxCommand.class);
+ when(finishTxCommand.groupType()).thenReturn(PartitionReplicationMessageGroup.GROUP_TYPE);
+ when(finishTxCommand.messageType()).thenReturn(Commands.FINISH_TX);
+
+ PrimaryReplicaChangeCommand primaryReplicaChangeCommand = mock(PrimaryReplicaChangeCommand.class);
+
+ // Checks for MvPartitionStorage.
+ listener.onWrite(List.of(
+ writeCommandCommandClosure(3, 1, updateCommand, updateCommandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(10, 1, updateCommand, updateCommandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(4, 1, writeIntentSwitchCommand, commandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(5, 1, safeTimeSyncCommand, commandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(6, 1, primaryReplicaChangeCommand, commandClosureResultCaptor, null)
+ ).iterator());
+
+ // Two storage runConsistently runs are expected: one for configuration application and another for primaryReplicaChangeCommand
+ // handling. Both comes from initial configuration preparation in @BeforeEach
Review Comment:
Legacy. Changed the comment with explanation of the expected methods' calls count.
##########
modules/partition-replicator/src/test/java/org/apache/ignite/internal/partition/replicator/raft/ZonePartitionRaftListenerTest.java:
##########
@@ -367,6 +405,151 @@ void processorsAreNotInitializedWithoutSnapshot(@Mock RaftTableProcessor tablePr
verify(tableProcessor, never()).initialize(any(), any(), anyLong(), anyLong());
}
+ @Test
+ void testSkipWriteCommandByAppliedIndex() {
+ mvPartitionStorage = spy(new TestMvPartitionStorage(PARTITION_ID));
+
+ PartitionListener tableProcessor = partitionListener(TABLE_ID);
+
+ listener.addTableProcessor(TABLE_ID, tableProcessor);
+ // Update(All)Command handling requires both information about raft group topology and the primary replica,
+ // thus onConfigurationCommited and primaryReplicaChangeCommand are called.
+ AtomicInteger raftIndex = new AtomicInteger();
+
+ long index = raftIndex.incrementAndGet();
+
+ listener.onConfigurationCommitted(
+ new RaftGroupConfiguration(
+ index,
+ 1,
+ List.of("foo"),
+ List.of("bar"),
+ null,
+ null
+ ),
+ index,
+ 1
+ );
+
+ PrimaryReplicaChangeCommand command = REPLICA_MESSAGES_FACTORY.primaryReplicaChangeCommand()
+ .primaryReplicaNodeName("primary")
+ .primaryReplicaNodeId(randomUUID())
+ .leaseStartTime(HybridTimestamp.MIN_VALUE.addPhysicalTime(1).longValue())
+ .build();
+
+ listener.onWrite(List.of(writeCommandCommandClosure(raftIndex.incrementAndGet(), 1, command, null, null)).iterator());
+
+ mvPartitionStorage.lastApplied(10L, 1L);
+
+ UpdateCommand updateCommand = mock(UpdateCommand.class);
+ when(updateCommand.tableId()).thenReturn(TABLE_ID);
+
+ WriteIntentSwitchCommand writeIntentSwitchCommand = mock(WriteIntentSwitchCommand.class);
+
+ SafeTimeSyncCommand safeTimeSyncCommand = mock(SafeTimeSyncCommand.class);
+
+ FinishTxCommand finishTxCommand = mock(FinishTxCommand.class);
+ when(finishTxCommand.groupType()).thenReturn(PartitionReplicationMessageGroup.GROUP_TYPE);
+ when(finishTxCommand.messageType()).thenReturn(Commands.FINISH_TX);
+
+ PrimaryReplicaChangeCommand primaryReplicaChangeCommand = mock(PrimaryReplicaChangeCommand.class);
+
+ // Checks for MvPartitionStorage.
+ listener.onWrite(List.of(
+ writeCommandCommandClosure(3, 1, updateCommand, updateCommandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(10, 1, updateCommand, updateCommandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(4, 1, writeIntentSwitchCommand, commandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(5, 1, safeTimeSyncCommand, commandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(6, 1, primaryReplicaChangeCommand, commandClosureResultCaptor, null)
+ ).iterator());
+
+ // Two storage runConsistently runs are expected: one for configuration application and another for primaryReplicaChangeCommand
+ // handling. Both comes from initial configuration preparation in @BeforeEach
+ verify(mvPartitionStorage, times(2)).runConsistently(any(WriteClosure.class));
+ verify(mvPartitionStorage, times(3)).lastApplied(anyLong(), anyLong()); // !! 3 vs 1
Review Comment:
My temporal note, removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-24770 Port tests related to partition lifecycle to PartitionReplicaLifecycleManager tests [ignite-3]
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #6004:
URL: https://github.com/apache/ignite-3/pull/6004#discussion_r2135609585
##########
modules/partition-replicator/src/test/java/org/apache/ignite/internal/partition/replicator/raft/ZonePartitionRaftListenerTest.java:
##########
@@ -367,6 +405,151 @@ void processorsAreNotInitializedWithoutSnapshot(@Mock RaftTableProcessor tablePr
verify(tableProcessor, never()).initialize(any(), any(), anyLong(), anyLong());
}
+ @Test
+ void testSkipWriteCommandByAppliedIndex() {
+ mvPartitionStorage = spy(new TestMvPartitionStorage(PARTITION_ID));
+
+ PartitionListener tableProcessor = partitionListener(TABLE_ID);
+
+ listener.addTableProcessor(TABLE_ID, tableProcessor);
+ // Update(All)Command handling requires both information about raft group topology and the primary replica,
+ // thus onConfigurationCommited and primaryReplicaChangeCommand are called.
+ AtomicInteger raftIndex = new AtomicInteger();
+
+ long index = raftIndex.incrementAndGet();
+
+ listener.onConfigurationCommitted(
+ new RaftGroupConfiguration(
+ index,
+ 1,
+ List.of("foo"),
+ List.of("bar"),
+ null,
+ null
+ ),
+ index,
+ 1
+ );
+
+ PrimaryReplicaChangeCommand command = REPLICA_MESSAGES_FACTORY.primaryReplicaChangeCommand()
+ .primaryReplicaNodeName("primary")
+ .primaryReplicaNodeId(randomUUID())
+ .leaseStartTime(HybridTimestamp.MIN_VALUE.addPhysicalTime(1).longValue())
+ .build();
+
+ listener.onWrite(List.of(writeCommandCommandClosure(raftIndex.incrementAndGet(), 1, command, null, null)).iterator());
+
+ mvPartitionStorage.lastApplied(10L, 1L);
+
+ UpdateCommand updateCommand = mock(UpdateCommand.class);
+ when(updateCommand.tableId()).thenReturn(TABLE_ID);
+
+ WriteIntentSwitchCommand writeIntentSwitchCommand = mock(WriteIntentSwitchCommand.class);
+
+ SafeTimeSyncCommand safeTimeSyncCommand = mock(SafeTimeSyncCommand.class);
+
+ FinishTxCommand finishTxCommand = mock(FinishTxCommand.class);
+ when(finishTxCommand.groupType()).thenReturn(PartitionReplicationMessageGroup.GROUP_TYPE);
+ when(finishTxCommand.messageType()).thenReturn(Commands.FINISH_TX);
+
+ PrimaryReplicaChangeCommand primaryReplicaChangeCommand = mock(PrimaryReplicaChangeCommand.class);
+
+ // Checks for MvPartitionStorage.
+ listener.onWrite(List.of(
+ writeCommandCommandClosure(3, 1, updateCommand, updateCommandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(10, 1, updateCommand, updateCommandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(4, 1, writeIntentSwitchCommand, commandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(5, 1, safeTimeSyncCommand, commandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(6, 1, primaryReplicaChangeCommand, commandClosureResultCaptor, null)
+ ).iterator());
+
+ // Two storage runConsistently runs are expected: one for configuration application and another for primaryReplicaChangeCommand
+ // handling. Both comes from initial configuration preparation in @BeforeEach
+ verify(mvPartitionStorage, times(2)).runConsistently(any(WriteClosure.class));
+ verify(mvPartitionStorage, times(3)).lastApplied(anyLong(), anyLong()); // !! 3 vs 1
Review Comment:
What does this comment mean?
##########
modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/ZonePartitionReplicaListenerTest.java:
##########
@@ -237,6 +277,10 @@ public class ZonePartitionReplicaListenerTest extends IgniteAbstractTest {
private final LockManager lockManager = lockManager();
+ @Captor
+ private ArgumentCaptor<Command> commandCaptor;
+
+
Review Comment:
Extra blank line?
##########
modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/ZonePartitionReplicaListenerTest.java:
##########
@@ -1275,6 +1323,492 @@ void exceptionIsReturnedIfTableProcessorIsAbsent() {
assertThat(ex.getMessage(), is("Table is already destroyed [tableId=1]"));
}
+ @Test
+ public void testTxStateReplicaRequestEmptyState() throws Exception {
+ doAnswer(invocation -> {
+ UUID txId = invocation.getArgument(5);
+
+ txManager.updateTxMeta(txId, old -> new TxStateMeta(
+ ABORTED,
+ localNode.id(),
+ commitPartitionId,
+ null,
+ null,
+ null
+ ));
+
+ return nullCompletedFuture();
+ }).when(txManager).finish(any(), any(), anyBoolean(), anyBoolean(), any(), any());
+
+ CompletableFuture<ReplicaResult> fut = zonePartitionReplicaListener.invoke(TX_MESSAGES_FACTORY.txStateCommitPartitionRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .txId(newTxId())
+ .enlistmentConsistencyToken(ANY_ENLISTMENT_CONSISTENCY_TOKEN)
+ .build(), randomUUID());
+
+ assertThat(fut, willSucceedFast());
+
+ TransactionMeta txMeta = (TransactionMeta) fut.get().result();
+
+ assertNotNull(txMeta);
+
+ assertEquals(ABORTED, txMeta.txState());
+ }
+
+ @Test
+ public void testTxStateReplicaRequestCommitState() throws Exception {
+ UUID txId = newTxId();
+
+ txStateStorage.putForRebalance(txId, new TxMeta(COMMITTED, singletonList(new EnlistedPartitionGroup(grpId)), clock.now()));
+
+ HybridTimestamp readTimestamp = clock.now();
+
+ CompletableFuture<ReplicaResult> fut = zonePartitionReplicaListener.invoke(TX_MESSAGES_FACTORY.txStateCommitPartitionRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .txId(txId)
+ .enlistmentConsistencyToken(ANY_ENLISTMENT_CONSISTENCY_TOKEN)
+ .build(), localNode.id());
+
+ assertThat(fut, willSucceedFast());
+
+ TransactionMeta txMeta = (TransactionMeta) fut.get().result();
+
+ assertNotNull(txMeta);
+ assertEquals(COMMITTED, txMeta.txState());
+ assertNotNull(txMeta.commitTimestamp());
+ assertTrue(readTimestamp.compareTo(txMeta.commitTimestamp()) > 0);
+ }
+
+ @Test
+ @WithSystemProperty(key = COLOCATION_FEATURE_FLAG, value = "true")
+ public void testWriteIntentOnPrimaryReplicaSingleUpdate() {
+ zonePartitionReplicaListener.addTableReplicaProcessor(TABLE_ID, mocked -> tableReplicaProcessor);
+
+ UUID txId = newTxId();
+ AtomicInteger counter = new AtomicInteger();
+
+ testWriteIntentOnPrimaryReplica(
+ txId,
+ () -> {
+ BinaryRow binaryRow = binaryRow(counter.getAndIncrement());
+
+ return TABLE_MESSAGES_FACTORY.readWriteSingleRowReplicaRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .tableId(TABLE_ID)
+ .transactionId(txId)
+ .requestType(RW_INSERT)
+ .schemaVersion(binaryRow.schemaVersion())
+ .binaryTuple(binaryRow.tupleSlice())
+ .enlistmentConsistencyToken(ANY_ENLISTMENT_CONSISTENCY_TOKEN)
+ .commitPartitionId(commitPartitionId())
+ .coordinatorId(localNode.id())
+ .timestamp(clock.now())
+ .build();
+ },
+ () -> checkRowInMvStorage(binaryRow(0), true)
+ );
+
+ cleanup(txId);
+
+ zonePartitionReplicaListener.removeTableReplicaProcessor(TABLE_ID);
+ }
+
+
+ @Test
+ @WithSystemProperty(key = COLOCATION_FEATURE_FLAG, value = "true")
+ public void testWriteIntentOnPrimaryReplicaUpdateAll() {
+ zonePartitionReplicaListener.addTableReplicaProcessor(TABLE_ID, mocked -> tableReplicaProcessor);
+
+ UUID txId = newTxId();
+ AtomicInteger counter = new AtomicInteger();
+
+ testWriteIntentOnPrimaryReplica(
+ txId,
+ () -> {
+ int cntr = counter.getAndIncrement();
+ BinaryRow binaryRow0 = binaryRow(cntr * 2);
+ BinaryRow binaryRow1 = binaryRow(cntr * 2 + 1);
+
+ return TABLE_MESSAGES_FACTORY.readWriteMultiRowReplicaRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .tableId(TABLE_ID)
+ .transactionId(txId)
+ .requestType(RW_UPSERT_ALL)
+ .schemaVersion(binaryRow0.schemaVersion())
+ .binaryTuples(asList(binaryRow0.tupleSlice(), binaryRow1.tupleSlice()))
+ .enlistmentConsistencyToken(ANY_ENLISTMENT_CONSISTENCY_TOKEN)
+ .commitPartitionId(commitPartitionId())
+ .coordinatorId(localNode.id())
+ .timestamp(clock.now())
+ .build();
+ },
+ () -> checkRowInMvStorage(binaryRow(0), true)
+ );
+
+ cleanup(txId);
+
+ zonePartitionReplicaListener.removeTableReplicaProcessor(TABLE_ID);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ @WithSystemProperty(key = COLOCATION_FEATURE_FLAG, value = "true")
+ void writeIntentSwitchForCompactedCatalogTimestampWorks(boolean commit) {
+ int earliestVersion = 999;
+ Catalog mockEarliestCatalog = mock(Catalog.class);
+ when(mockEarliestCatalog.version()).thenReturn(earliestVersion);
+
+ UUID txId = newTxId();
+ HybridTimestamp beginTs = beginTimestamp(txId);
+ HybridTimestamp commitTs = clock.now();
+
+ HybridTimestamp reliableCatalogVersionTs = commit ? commitTs : beginTs;
+ when(catalogService.activeCatalog(reliableCatalogVersionTs.longValue())).thenThrow(new CatalogNotFoundException("Oops"));
+ when(catalogService.earliestCatalog()).thenReturn(mockEarliestCatalog);
+
+ zonePartitionReplicaListener.addTableReplicaProcessor(TABLE_ID, mocked -> tableReplicaProcessor);
+
+ CompletableFuture<ReplicaResult> invokeFuture = zonePartitionReplicaListener.invoke(
+ TX_MESSAGES_FACTORY.writeIntentSwitchReplicaRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .tableIds(Set.of(grpId.tableId()))
+ .txId(txId)
+ .commit(commit)
+ .commitTimestamp(commit ? commitTs : null)
+ .build(),
+ localNode.id()
+ );
+
+ assertThat(invokeFuture, willCompleteSuccessfully());
+ assertThat(invokeFuture.join().applyResult().replicationFuture(), willCompleteSuccessfully());
+
+ verify(mockRaftClient).run(commandCaptor.capture());
+ WriteIntentSwitchCommand command = (WriteIntentSwitchCommand) commandCaptor.getValue();
+
+ assertThat(command.requiredCatalogVersion(), is(earliestVersion));
+
+ zonePartitionReplicaListener.removeTableReplicaProcessor(TABLE_ID);
+ }
+
+
Review Comment:
Extra blank line?
##########
modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/ZonePartitionReplicaListenerTest.java:
##########
@@ -565,6 +609,10 @@ public ClusterNode getByConsistentId(String consistentId) {
placementDriver = spy(new TestPlacementDriver(localNode));
+ ZonePartitionId zonePartitionId = new ZonePartitionId(tableDescriptor.zoneId(), PART_ID);
+
+ FailureManager failureHandler = new NoOpFailureManager();
Review Comment:
It's not a handler, rather a manager. `FailureHandler` is another entity
##########
modules/partition-replicator/src/test/java/org/apache/ignite/internal/partition/replicator/raft/ZonePartitionRaftListenerTest.java:
##########
@@ -367,6 +405,151 @@ void processorsAreNotInitializedWithoutSnapshot(@Mock RaftTableProcessor tablePr
verify(tableProcessor, never()).initialize(any(), any(), anyLong(), anyLong());
}
+ @Test
+ void testSkipWriteCommandByAppliedIndex() {
+ mvPartitionStorage = spy(new TestMvPartitionStorage(PARTITION_ID));
+
+ PartitionListener tableProcessor = partitionListener(TABLE_ID);
+
+ listener.addTableProcessor(TABLE_ID, tableProcessor);
+ // Update(All)Command handling requires both information about raft group topology and the primary replica,
+ // thus onConfigurationCommited and primaryReplicaChangeCommand are called.
+ AtomicInteger raftIndex = new AtomicInteger();
+
+ long index = raftIndex.incrementAndGet();
+
+ listener.onConfigurationCommitted(
+ new RaftGroupConfiguration(
+ index,
+ 1,
+ List.of("foo"),
+ List.of("bar"),
+ null,
+ null
+ ),
+ index,
+ 1
+ );
+
+ PrimaryReplicaChangeCommand command = REPLICA_MESSAGES_FACTORY.primaryReplicaChangeCommand()
+ .primaryReplicaNodeName("primary")
+ .primaryReplicaNodeId(randomUUID())
+ .leaseStartTime(HybridTimestamp.MIN_VALUE.addPhysicalTime(1).longValue())
+ .build();
+
+ listener.onWrite(List.of(writeCommandCommandClosure(raftIndex.incrementAndGet(), 1, command, null, null)).iterator());
+
+ mvPartitionStorage.lastApplied(10L, 1L);
+
+ UpdateCommand updateCommand = mock(UpdateCommand.class);
+ when(updateCommand.tableId()).thenReturn(TABLE_ID);
+
+ WriteIntentSwitchCommand writeIntentSwitchCommand = mock(WriteIntentSwitchCommand.class);
+
+ SafeTimeSyncCommand safeTimeSyncCommand = mock(SafeTimeSyncCommand.class);
+
+ FinishTxCommand finishTxCommand = mock(FinishTxCommand.class);
+ when(finishTxCommand.groupType()).thenReturn(PartitionReplicationMessageGroup.GROUP_TYPE);
+ when(finishTxCommand.messageType()).thenReturn(Commands.FINISH_TX);
+
+ PrimaryReplicaChangeCommand primaryReplicaChangeCommand = mock(PrimaryReplicaChangeCommand.class);
+
+ // Checks for MvPartitionStorage.
+ listener.onWrite(List.of(
+ writeCommandCommandClosure(3, 1, updateCommand, updateCommandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(10, 1, updateCommand, updateCommandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(4, 1, writeIntentSwitchCommand, commandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(5, 1, safeTimeSyncCommand, commandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(6, 1, primaryReplicaChangeCommand, commandClosureResultCaptor, null)
+ ).iterator());
+
+ // Two storage runConsistently runs are expected: one for configuration application and another for primaryReplicaChangeCommand
+ // handling. Both comes from initial configuration preparation in @BeforeEach
+ verify(mvPartitionStorage, times(2)).runConsistently(any(WriteClosure.class));
+ verify(mvPartitionStorage, times(3)).lastApplied(anyLong(), anyLong()); // !! 3 vs 1
+
+ List<UpdateCommandResult> allValues = updateCommandClosureResultCaptor.getAllValues();
+ assertThat(allValues, containsInAnyOrder(new Throwable[]{null, null}));
+ assertThat(commandClosureResultCaptor.getAllValues(), containsInAnyOrder(new Throwable[]{null, null, null}));
+
+ // Checks for TxStateStorage.
+ mvPartitionStorage.lastApplied(1L, 1L);
+ txStatePartitionStorage.lastApplied(10L, 2L);
+
+ commandClosureResultCaptor = ArgumentCaptor.forClass(Throwable.class);
+
+ listener.onWrite(List.of(
+ writeCommandCommandClosure(2, 1, finishTxCommand, commandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(10, 1, finishTxCommand, commandClosureResultCaptor, clock.now())
+ ).iterator());
+
+ verify(txStatePartitionStorage, never())
+ .compareAndSet(any(UUID.class), any(TxState.class), any(TxMeta.class), anyLong(), anyLong());
+ verify(txStatePartitionStorage, times(1)).lastApplied(anyLong(), anyLong());
+
+ assertThat(commandClosureResultCaptor.getAllValues(), containsInAnyOrder(new Throwable[]{null, null}));
+
+ listener.removeTableProcessor(TABLE_ID);
+ }
+
+
+ @Test
+ void updatesLastAppliedForFinishTxCommands() {
+ safeTimeClock.update(clock.now(), null);
+
+ FinishTxCommand command = PARTITION_REPLICATION_MESSAGES_FACTORY.finishTxCommand()
+ .txId(TestTransactionIds.newTransactionId())
+ .initiatorTime(clock.now())
+ .partitions(List.of())
+ .build();
+
+ listener.onWrite(List.of(
+ writeCommandCommandClosure(3, 2, command)
+ ).iterator());
+
+ assertThat(txStatePartitionStorage.lastAppliedIndex(), is(3L));
+ assertThat(txStatePartitionStorage.lastAppliedTerm(), is(2L));
+ }
+
+ private CommandClosure<WriteCommand> writeCommandCommandClosure(
Review Comment:
```suggestion
private CommandClosure<WriteCommand> writeCommandClosure(
```
##########
modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/ZonePartitionReplicaListenerTest.java:
##########
@@ -322,7 +366,7 @@ public class ZonePartitionReplicaListenerTest extends IgniteAbstractTest {
private TopologyService topologySrv;
@Mock
- private PendingComparableValuesTracker<HybridTimestamp, Void> safeTimeClock;
+ private SafeTimeValuesTracker safeTimeClock;
Review Comment:
It's not a clock
##########
modules/partition-replicator/src/test/java/org/apache/ignite/internal/partition/replicator/raft/ZonePartitionRaftListenerTest.java:
##########
@@ -367,6 +405,151 @@ void processorsAreNotInitializedWithoutSnapshot(@Mock RaftTableProcessor tablePr
verify(tableProcessor, never()).initialize(any(), any(), anyLong(), anyLong());
}
+ @Test
+ void testSkipWriteCommandByAppliedIndex() {
+ mvPartitionStorage = spy(new TestMvPartitionStorage(PARTITION_ID));
+
+ PartitionListener tableProcessor = partitionListener(TABLE_ID);
+
+ listener.addTableProcessor(TABLE_ID, tableProcessor);
+ // Update(All)Command handling requires both information about raft group topology and the primary replica,
+ // thus onConfigurationCommited and primaryReplicaChangeCommand are called.
+ AtomicInteger raftIndex = new AtomicInteger();
+
+ long index = raftIndex.incrementAndGet();
+
+ listener.onConfigurationCommitted(
+ new RaftGroupConfiguration(
+ index,
+ 1,
+ List.of("foo"),
+ List.of("bar"),
+ null,
+ null
+ ),
+ index,
+ 1
+ );
+
+ PrimaryReplicaChangeCommand command = REPLICA_MESSAGES_FACTORY.primaryReplicaChangeCommand()
+ .primaryReplicaNodeName("primary")
+ .primaryReplicaNodeId(randomUUID())
+ .leaseStartTime(HybridTimestamp.MIN_VALUE.addPhysicalTime(1).longValue())
+ .build();
+
+ listener.onWrite(List.of(writeCommandCommandClosure(raftIndex.incrementAndGet(), 1, command, null, null)).iterator());
+
+ mvPartitionStorage.lastApplied(10L, 1L);
+
+ UpdateCommand updateCommand = mock(UpdateCommand.class);
+ when(updateCommand.tableId()).thenReturn(TABLE_ID);
+
+ WriteIntentSwitchCommand writeIntentSwitchCommand = mock(WriteIntentSwitchCommand.class);
+
+ SafeTimeSyncCommand safeTimeSyncCommand = mock(SafeTimeSyncCommand.class);
+
+ FinishTxCommand finishTxCommand = mock(FinishTxCommand.class);
+ when(finishTxCommand.groupType()).thenReturn(PartitionReplicationMessageGroup.GROUP_TYPE);
+ when(finishTxCommand.messageType()).thenReturn(Commands.FINISH_TX);
+
+ PrimaryReplicaChangeCommand primaryReplicaChangeCommand = mock(PrimaryReplicaChangeCommand.class);
+
+ // Checks for MvPartitionStorage.
+ listener.onWrite(List.of(
+ writeCommandCommandClosure(3, 1, updateCommand, updateCommandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(10, 1, updateCommand, updateCommandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(4, 1, writeIntentSwitchCommand, commandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(5, 1, safeTimeSyncCommand, commandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(6, 1, primaryReplicaChangeCommand, commandClosureResultCaptor, null)
+ ).iterator());
+
+ // Two storage runConsistently runs are expected: one for configuration application and another for primaryReplicaChangeCommand
+ // handling. Both comes from initial configuration preparation in @BeforeEach
Review Comment:
Do they still come from `@BeforeEach`?
##########
modules/partition-replicator/src/test/java/org/apache/ignite/internal/partition/replicator/raft/ZonePartitionRaftListenerTest.java:
##########
@@ -112,6 +145,11 @@ class ZonePartitionRaftListenerTest extends BaseIgniteAbstractTest {
@InjectExecutorService
private ExecutorService executor;
+ @Mock
+ private SafeTimeValuesTracker safeTimeClock;
Review Comment:
It is not a clock, so the name seems to be a bit misleading
##########
modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/ZonePartitionReplicaListenerTest.java:
##########
@@ -1275,6 +1323,492 @@ void exceptionIsReturnedIfTableProcessorIsAbsent() {
assertThat(ex.getMessage(), is("Table is already destroyed [tableId=1]"));
}
+ @Test
+ public void testTxStateReplicaRequestEmptyState() throws Exception {
+ doAnswer(invocation -> {
+ UUID txId = invocation.getArgument(5);
+
+ txManager.updateTxMeta(txId, old -> new TxStateMeta(
+ ABORTED,
+ localNode.id(),
+ commitPartitionId,
+ null,
+ null,
+ null
+ ));
+
+ return nullCompletedFuture();
+ }).when(txManager).finish(any(), any(), anyBoolean(), anyBoolean(), any(), any());
+
+ CompletableFuture<ReplicaResult> fut = zonePartitionReplicaListener.invoke(TX_MESSAGES_FACTORY.txStateCommitPartitionRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .txId(newTxId())
+ .enlistmentConsistencyToken(ANY_ENLISTMENT_CONSISTENCY_TOKEN)
+ .build(), randomUUID());
+
+ assertThat(fut, willSucceedFast());
+
+ TransactionMeta txMeta = (TransactionMeta) fut.get().result();
+
+ assertNotNull(txMeta);
+
+ assertEquals(ABORTED, txMeta.txState());
+ }
+
+ @Test
+ public void testTxStateReplicaRequestCommitState() throws Exception {
+ UUID txId = newTxId();
+
+ txStateStorage.putForRebalance(txId, new TxMeta(COMMITTED, singletonList(new EnlistedPartitionGroup(grpId)), clock.now()));
+
+ HybridTimestamp readTimestamp = clock.now();
+
+ CompletableFuture<ReplicaResult> fut = zonePartitionReplicaListener.invoke(TX_MESSAGES_FACTORY.txStateCommitPartitionRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .txId(txId)
+ .enlistmentConsistencyToken(ANY_ENLISTMENT_CONSISTENCY_TOKEN)
+ .build(), localNode.id());
+
+ assertThat(fut, willSucceedFast());
+
+ TransactionMeta txMeta = (TransactionMeta) fut.get().result();
+
+ assertNotNull(txMeta);
+ assertEquals(COMMITTED, txMeta.txState());
+ assertNotNull(txMeta.commitTimestamp());
+ assertTrue(readTimestamp.compareTo(txMeta.commitTimestamp()) > 0);
+ }
+
+ @Test
+ @WithSystemProperty(key = COLOCATION_FEATURE_FLAG, value = "true")
+ public void testWriteIntentOnPrimaryReplicaSingleUpdate() {
+ zonePartitionReplicaListener.addTableReplicaProcessor(TABLE_ID, mocked -> tableReplicaProcessor);
+
+ UUID txId = newTxId();
+ AtomicInteger counter = new AtomicInteger();
+
+ testWriteIntentOnPrimaryReplica(
+ txId,
+ () -> {
+ BinaryRow binaryRow = binaryRow(counter.getAndIncrement());
+
+ return TABLE_MESSAGES_FACTORY.readWriteSingleRowReplicaRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .tableId(TABLE_ID)
+ .transactionId(txId)
+ .requestType(RW_INSERT)
+ .schemaVersion(binaryRow.schemaVersion())
+ .binaryTuple(binaryRow.tupleSlice())
+ .enlistmentConsistencyToken(ANY_ENLISTMENT_CONSISTENCY_TOKEN)
+ .commitPartitionId(commitPartitionId())
+ .coordinatorId(localNode.id())
+ .timestamp(clock.now())
+ .build();
+ },
+ () -> checkRowInMvStorage(binaryRow(0), true)
+ );
+
+ cleanup(txId);
+
+ zonePartitionReplicaListener.removeTableReplicaProcessor(TABLE_ID);
+ }
+
+
+ @Test
+ @WithSystemProperty(key = COLOCATION_FEATURE_FLAG, value = "true")
+ public void testWriteIntentOnPrimaryReplicaUpdateAll() {
+ zonePartitionReplicaListener.addTableReplicaProcessor(TABLE_ID, mocked -> tableReplicaProcessor);
+
+ UUID txId = newTxId();
+ AtomicInteger counter = new AtomicInteger();
+
+ testWriteIntentOnPrimaryReplica(
+ txId,
+ () -> {
+ int cntr = counter.getAndIncrement();
+ BinaryRow binaryRow0 = binaryRow(cntr * 2);
+ BinaryRow binaryRow1 = binaryRow(cntr * 2 + 1);
+
+ return TABLE_MESSAGES_FACTORY.readWriteMultiRowReplicaRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .tableId(TABLE_ID)
+ .transactionId(txId)
+ .requestType(RW_UPSERT_ALL)
+ .schemaVersion(binaryRow0.schemaVersion())
+ .binaryTuples(asList(binaryRow0.tupleSlice(), binaryRow1.tupleSlice()))
+ .enlistmentConsistencyToken(ANY_ENLISTMENT_CONSISTENCY_TOKEN)
+ .commitPartitionId(commitPartitionId())
+ .coordinatorId(localNode.id())
+ .timestamp(clock.now())
+ .build();
+ },
+ () -> checkRowInMvStorage(binaryRow(0), true)
+ );
+
+ cleanup(txId);
+
+ zonePartitionReplicaListener.removeTableReplicaProcessor(TABLE_ID);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ @WithSystemProperty(key = COLOCATION_FEATURE_FLAG, value = "true")
+ void writeIntentSwitchForCompactedCatalogTimestampWorks(boolean commit) {
+ int earliestVersion = 999;
+ Catalog mockEarliestCatalog = mock(Catalog.class);
+ when(mockEarliestCatalog.version()).thenReturn(earliestVersion);
+
+ UUID txId = newTxId();
+ HybridTimestamp beginTs = beginTimestamp(txId);
+ HybridTimestamp commitTs = clock.now();
+
+ HybridTimestamp reliableCatalogVersionTs = commit ? commitTs : beginTs;
+ when(catalogService.activeCatalog(reliableCatalogVersionTs.longValue())).thenThrow(new CatalogNotFoundException("Oops"));
+ when(catalogService.earliestCatalog()).thenReturn(mockEarliestCatalog);
+
+ zonePartitionReplicaListener.addTableReplicaProcessor(TABLE_ID, mocked -> tableReplicaProcessor);
+
+ CompletableFuture<ReplicaResult> invokeFuture = zonePartitionReplicaListener.invoke(
+ TX_MESSAGES_FACTORY.writeIntentSwitchReplicaRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .tableIds(Set.of(grpId.tableId()))
+ .txId(txId)
+ .commit(commit)
+ .commitTimestamp(commit ? commitTs : null)
+ .build(),
+ localNode.id()
+ );
+
+ assertThat(invokeFuture, willCompleteSuccessfully());
+ assertThat(invokeFuture.join().applyResult().replicationFuture(), willCompleteSuccessfully());
+
+ verify(mockRaftClient).run(commandCaptor.capture());
+ WriteIntentSwitchCommand command = (WriteIntentSwitchCommand) commandCaptor.getValue();
+
+ assertThat(command.requiredCatalogVersion(), is(earliestVersion));
+
+ zonePartitionReplicaListener.removeTableReplicaProcessor(TABLE_ID);
+ }
+
+
+ @Test
+ public void abortsSuccessfully() {
+ AtomicReference<Boolean> committed = interceptFinishTxCommand();
+
+ CompletableFuture<?> future = beginAndAbortTx();
+
+ assertThat(future, willSucceedFast());
+
+ assertThat(committed.get(), is(false));
+ }
+
+ @Test
+ public void commitsOnSameSchemaSuccessfully() {
+ when(validationSchemasSource.tableSchemaVersionsBetween(anyInt(), any(), any(HybridTimestamp.class)))
+ .thenReturn(List.of(
+ tableSchema(CURRENT_SCHEMA_VERSION, List.of(nullableColumn("col")))
+ ));
+
+ AtomicReference<Boolean> committed = interceptFinishTxCommand();
+
+ CompletableFuture<?> future = beginAndCommitTx();
+
+ assertThat(future, willSucceedFast());
+
+ assertThat(committed.get(), is(true));
+ }
+
+
Review Comment:
And here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-24770 Copy PartitionReplicaListenerTest logic to test ZonePartitionReplicaListener [ignite-3]
Posted by "JAkutenshi (via GitHub)" <gi...@apache.org>.
JAkutenshi commented on code in PR #6004:
URL: https://github.com/apache/ignite-3/pull/6004#discussion_r2137575078
##########
modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/ZonePartitionReplicaListenerTest.java:
##########
@@ -1275,6 +1323,492 @@ void exceptionIsReturnedIfTableProcessorIsAbsent() {
assertThat(ex.getMessage(), is("Table is already destroyed [tableId=1]"));
}
+ @Test
+ public void testTxStateReplicaRequestEmptyState() throws Exception {
+ doAnswer(invocation -> {
+ UUID txId = invocation.getArgument(5);
+
+ txManager.updateTxMeta(txId, old -> new TxStateMeta(
+ ABORTED,
+ localNode.id(),
+ commitPartitionId,
+ null,
+ null,
+ null
+ ));
+
+ return nullCompletedFuture();
+ }).when(txManager).finish(any(), any(), anyBoolean(), anyBoolean(), any(), any());
+
+ CompletableFuture<ReplicaResult> fut = zonePartitionReplicaListener.invoke(TX_MESSAGES_FACTORY.txStateCommitPartitionRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .txId(newTxId())
+ .enlistmentConsistencyToken(ANY_ENLISTMENT_CONSISTENCY_TOKEN)
+ .build(), randomUUID());
+
+ assertThat(fut, willSucceedFast());
+
+ TransactionMeta txMeta = (TransactionMeta) fut.get().result();
+
+ assertNotNull(txMeta);
+
+ assertEquals(ABORTED, txMeta.txState());
+ }
+
+ @Test
+ public void testTxStateReplicaRequestCommitState() throws Exception {
+ UUID txId = newTxId();
+
+ txStateStorage.putForRebalance(txId, new TxMeta(COMMITTED, singletonList(new EnlistedPartitionGroup(grpId)), clock.now()));
+
+ HybridTimestamp readTimestamp = clock.now();
+
+ CompletableFuture<ReplicaResult> fut = zonePartitionReplicaListener.invoke(TX_MESSAGES_FACTORY.txStateCommitPartitionRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .txId(txId)
+ .enlistmentConsistencyToken(ANY_ENLISTMENT_CONSISTENCY_TOKEN)
+ .build(), localNode.id());
+
+ assertThat(fut, willSucceedFast());
+
+ TransactionMeta txMeta = (TransactionMeta) fut.get().result();
+
+ assertNotNull(txMeta);
+ assertEquals(COMMITTED, txMeta.txState());
+ assertNotNull(txMeta.commitTimestamp());
+ assertTrue(readTimestamp.compareTo(txMeta.commitTimestamp()) > 0);
+ }
+
+ @Test
+ @WithSystemProperty(key = COLOCATION_FEATURE_FLAG, value = "true")
+ public void testWriteIntentOnPrimaryReplicaSingleUpdate() {
+ zonePartitionReplicaListener.addTableReplicaProcessor(TABLE_ID, mocked -> tableReplicaProcessor);
+
+ UUID txId = newTxId();
+ AtomicInteger counter = new AtomicInteger();
+
+ testWriteIntentOnPrimaryReplica(
+ txId,
+ () -> {
+ BinaryRow binaryRow = binaryRow(counter.getAndIncrement());
+
+ return TABLE_MESSAGES_FACTORY.readWriteSingleRowReplicaRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .tableId(TABLE_ID)
+ .transactionId(txId)
+ .requestType(RW_INSERT)
+ .schemaVersion(binaryRow.schemaVersion())
+ .binaryTuple(binaryRow.tupleSlice())
+ .enlistmentConsistencyToken(ANY_ENLISTMENT_CONSISTENCY_TOKEN)
+ .commitPartitionId(commitPartitionId())
+ .coordinatorId(localNode.id())
+ .timestamp(clock.now())
+ .build();
+ },
+ () -> checkRowInMvStorage(binaryRow(0), true)
+ );
+
+ cleanup(txId);
+
+ zonePartitionReplicaListener.removeTableReplicaProcessor(TABLE_ID);
+ }
+
+
+ @Test
+ @WithSystemProperty(key = COLOCATION_FEATURE_FLAG, value = "true")
+ public void testWriteIntentOnPrimaryReplicaUpdateAll() {
+ zonePartitionReplicaListener.addTableReplicaProcessor(TABLE_ID, mocked -> tableReplicaProcessor);
+
+ UUID txId = newTxId();
+ AtomicInteger counter = new AtomicInteger();
+
+ testWriteIntentOnPrimaryReplica(
+ txId,
+ () -> {
+ int cntr = counter.getAndIncrement();
+ BinaryRow binaryRow0 = binaryRow(cntr * 2);
+ BinaryRow binaryRow1 = binaryRow(cntr * 2 + 1);
+
+ return TABLE_MESSAGES_FACTORY.readWriteMultiRowReplicaRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .tableId(TABLE_ID)
+ .transactionId(txId)
+ .requestType(RW_UPSERT_ALL)
+ .schemaVersion(binaryRow0.schemaVersion())
+ .binaryTuples(asList(binaryRow0.tupleSlice(), binaryRow1.tupleSlice()))
+ .enlistmentConsistencyToken(ANY_ENLISTMENT_CONSISTENCY_TOKEN)
+ .commitPartitionId(commitPartitionId())
+ .coordinatorId(localNode.id())
+ .timestamp(clock.now())
+ .build();
+ },
+ () -> checkRowInMvStorage(binaryRow(0), true)
+ );
+
+ cleanup(txId);
+
+ zonePartitionReplicaListener.removeTableReplicaProcessor(TABLE_ID);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ @WithSystemProperty(key = COLOCATION_FEATURE_FLAG, value = "true")
+ void writeIntentSwitchForCompactedCatalogTimestampWorks(boolean commit) {
+ int earliestVersion = 999;
+ Catalog mockEarliestCatalog = mock(Catalog.class);
+ when(mockEarliestCatalog.version()).thenReturn(earliestVersion);
+
+ UUID txId = newTxId();
+ HybridTimestamp beginTs = beginTimestamp(txId);
+ HybridTimestamp commitTs = clock.now();
+
+ HybridTimestamp reliableCatalogVersionTs = commit ? commitTs : beginTs;
+ when(catalogService.activeCatalog(reliableCatalogVersionTs.longValue())).thenThrow(new CatalogNotFoundException("Oops"));
+ when(catalogService.earliestCatalog()).thenReturn(mockEarliestCatalog);
+
+ zonePartitionReplicaListener.addTableReplicaProcessor(TABLE_ID, mocked -> tableReplicaProcessor);
+
+ CompletableFuture<ReplicaResult> invokeFuture = zonePartitionReplicaListener.invoke(
+ TX_MESSAGES_FACTORY.writeIntentSwitchReplicaRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .tableIds(Set.of(grpId.tableId()))
+ .txId(txId)
+ .commit(commit)
+ .commitTimestamp(commit ? commitTs : null)
+ .build(),
+ localNode.id()
+ );
+
+ assertThat(invokeFuture, willCompleteSuccessfully());
+ assertThat(invokeFuture.join().applyResult().replicationFuture(), willCompleteSuccessfully());
+
+ verify(mockRaftClient).run(commandCaptor.capture());
+ WriteIntentSwitchCommand command = (WriteIntentSwitchCommand) commandCaptor.getValue();
+
+ assertThat(command.requiredCatalogVersion(), is(earliestVersion));
+
+ zonePartitionReplicaListener.removeTableReplicaProcessor(TABLE_ID);
+ }
+
+
+ @Test
+ public void abortsSuccessfully() {
+ AtomicReference<Boolean> committed = interceptFinishTxCommand();
+
+ CompletableFuture<?> future = beginAndAbortTx();
+
+ assertThat(future, willSucceedFast());
+
+ assertThat(committed.get(), is(false));
+ }
+
+ @Test
+ public void commitsOnSameSchemaSuccessfully() {
+ when(validationSchemasSource.tableSchemaVersionsBetween(anyInt(), any(), any(HybridTimestamp.class)))
+ .thenReturn(List.of(
+ tableSchema(CURRENT_SCHEMA_VERSION, List.of(nullableColumn("col")))
+ ));
+
+ AtomicReference<Boolean> committed = interceptFinishTxCommand();
+
+ CompletableFuture<?> future = beginAndCommitTx();
+
+ assertThat(future, willSucceedFast());
+
+ assertThat(committed.get(), is(true));
+ }
+
+
Review Comment:
And removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-24770 Copy PartitionReplicaListenerTest logic to test ZonePartitionReplicaListener [ignite-3]
Posted by "JAkutenshi (via GitHub)" <gi...@apache.org>.
JAkutenshi commented on code in PR #6004:
URL: https://github.com/apache/ignite-3/pull/6004#discussion_r2137574561
##########
modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/ZonePartitionReplicaListenerTest.java:
##########
@@ -565,6 +609,10 @@ public ClusterNode getByConsistentId(String consistentId) {
placementDriver = spy(new TestPlacementDriver(localNode));
+ ZonePartitionId zonePartitionId = new ZonePartitionId(tableDescriptor.zoneId(), PART_ID);
+
+ FailureManager failureHandler = new NoOpFailureManager();
Review Comment:
Fair, renamed.
##########
modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/ZonePartitionReplicaListenerTest.java:
##########
@@ -1275,6 +1323,492 @@ void exceptionIsReturnedIfTableProcessorIsAbsent() {
assertThat(ex.getMessage(), is("Table is already destroyed [tableId=1]"));
}
+ @Test
+ public void testTxStateReplicaRequestEmptyState() throws Exception {
+ doAnswer(invocation -> {
+ UUID txId = invocation.getArgument(5);
+
+ txManager.updateTxMeta(txId, old -> new TxStateMeta(
+ ABORTED,
+ localNode.id(),
+ commitPartitionId,
+ null,
+ null,
+ null
+ ));
+
+ return nullCompletedFuture();
+ }).when(txManager).finish(any(), any(), anyBoolean(), anyBoolean(), any(), any());
+
+ CompletableFuture<ReplicaResult> fut = zonePartitionReplicaListener.invoke(TX_MESSAGES_FACTORY.txStateCommitPartitionRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .txId(newTxId())
+ .enlistmentConsistencyToken(ANY_ENLISTMENT_CONSISTENCY_TOKEN)
+ .build(), randomUUID());
+
+ assertThat(fut, willSucceedFast());
+
+ TransactionMeta txMeta = (TransactionMeta) fut.get().result();
+
+ assertNotNull(txMeta);
+
+ assertEquals(ABORTED, txMeta.txState());
+ }
+
+ @Test
+ public void testTxStateReplicaRequestCommitState() throws Exception {
+ UUID txId = newTxId();
+
+ txStateStorage.putForRebalance(txId, new TxMeta(COMMITTED, singletonList(new EnlistedPartitionGroup(grpId)), clock.now()));
+
+ HybridTimestamp readTimestamp = clock.now();
+
+ CompletableFuture<ReplicaResult> fut = zonePartitionReplicaListener.invoke(TX_MESSAGES_FACTORY.txStateCommitPartitionRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .txId(txId)
+ .enlistmentConsistencyToken(ANY_ENLISTMENT_CONSISTENCY_TOKEN)
+ .build(), localNode.id());
+
+ assertThat(fut, willSucceedFast());
+
+ TransactionMeta txMeta = (TransactionMeta) fut.get().result();
+
+ assertNotNull(txMeta);
+ assertEquals(COMMITTED, txMeta.txState());
+ assertNotNull(txMeta.commitTimestamp());
+ assertTrue(readTimestamp.compareTo(txMeta.commitTimestamp()) > 0);
+ }
+
+ @Test
+ @WithSystemProperty(key = COLOCATION_FEATURE_FLAG, value = "true")
+ public void testWriteIntentOnPrimaryReplicaSingleUpdate() {
+ zonePartitionReplicaListener.addTableReplicaProcessor(TABLE_ID, mocked -> tableReplicaProcessor);
+
+ UUID txId = newTxId();
+ AtomicInteger counter = new AtomicInteger();
+
+ testWriteIntentOnPrimaryReplica(
+ txId,
+ () -> {
+ BinaryRow binaryRow = binaryRow(counter.getAndIncrement());
+
+ return TABLE_MESSAGES_FACTORY.readWriteSingleRowReplicaRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .tableId(TABLE_ID)
+ .transactionId(txId)
+ .requestType(RW_INSERT)
+ .schemaVersion(binaryRow.schemaVersion())
+ .binaryTuple(binaryRow.tupleSlice())
+ .enlistmentConsistencyToken(ANY_ENLISTMENT_CONSISTENCY_TOKEN)
+ .commitPartitionId(commitPartitionId())
+ .coordinatorId(localNode.id())
+ .timestamp(clock.now())
+ .build();
+ },
+ () -> checkRowInMvStorage(binaryRow(0), true)
+ );
+
+ cleanup(txId);
+
+ zonePartitionReplicaListener.removeTableReplicaProcessor(TABLE_ID);
+ }
+
+
+ @Test
+ @WithSystemProperty(key = COLOCATION_FEATURE_FLAG, value = "true")
+ public void testWriteIntentOnPrimaryReplicaUpdateAll() {
+ zonePartitionReplicaListener.addTableReplicaProcessor(TABLE_ID, mocked -> tableReplicaProcessor);
+
+ UUID txId = newTxId();
+ AtomicInteger counter = new AtomicInteger();
+
+ testWriteIntentOnPrimaryReplica(
+ txId,
+ () -> {
+ int cntr = counter.getAndIncrement();
+ BinaryRow binaryRow0 = binaryRow(cntr * 2);
+ BinaryRow binaryRow1 = binaryRow(cntr * 2 + 1);
+
+ return TABLE_MESSAGES_FACTORY.readWriteMultiRowReplicaRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .tableId(TABLE_ID)
+ .transactionId(txId)
+ .requestType(RW_UPSERT_ALL)
+ .schemaVersion(binaryRow0.schemaVersion())
+ .binaryTuples(asList(binaryRow0.tupleSlice(), binaryRow1.tupleSlice()))
+ .enlistmentConsistencyToken(ANY_ENLISTMENT_CONSISTENCY_TOKEN)
+ .commitPartitionId(commitPartitionId())
+ .coordinatorId(localNode.id())
+ .timestamp(clock.now())
+ .build();
+ },
+ () -> checkRowInMvStorage(binaryRow(0), true)
+ );
+
+ cleanup(txId);
+
+ zonePartitionReplicaListener.removeTableReplicaProcessor(TABLE_ID);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ @WithSystemProperty(key = COLOCATION_FEATURE_FLAG, value = "true")
+ void writeIntentSwitchForCompactedCatalogTimestampWorks(boolean commit) {
+ int earliestVersion = 999;
+ Catalog mockEarliestCatalog = mock(Catalog.class);
+ when(mockEarliestCatalog.version()).thenReturn(earliestVersion);
+
+ UUID txId = newTxId();
+ HybridTimestamp beginTs = beginTimestamp(txId);
+ HybridTimestamp commitTs = clock.now();
+
+ HybridTimestamp reliableCatalogVersionTs = commit ? commitTs : beginTs;
+ when(catalogService.activeCatalog(reliableCatalogVersionTs.longValue())).thenThrow(new CatalogNotFoundException("Oops"));
+ when(catalogService.earliestCatalog()).thenReturn(mockEarliestCatalog);
+
+ zonePartitionReplicaListener.addTableReplicaProcessor(TABLE_ID, mocked -> tableReplicaProcessor);
+
+ CompletableFuture<ReplicaResult> invokeFuture = zonePartitionReplicaListener.invoke(
+ TX_MESSAGES_FACTORY.writeIntentSwitchReplicaRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .tableIds(Set.of(grpId.tableId()))
+ .txId(txId)
+ .commit(commit)
+ .commitTimestamp(commit ? commitTs : null)
+ .build(),
+ localNode.id()
+ );
+
+ assertThat(invokeFuture, willCompleteSuccessfully());
+ assertThat(invokeFuture.join().applyResult().replicationFuture(), willCompleteSuccessfully());
+
+ verify(mockRaftClient).run(commandCaptor.capture());
+ WriteIntentSwitchCommand command = (WriteIntentSwitchCommand) commandCaptor.getValue();
+
+ assertThat(command.requiredCatalogVersion(), is(earliestVersion));
+
+ zonePartitionReplicaListener.removeTableReplicaProcessor(TABLE_ID);
+ }
+
+
Review Comment:
Removed.
##########
modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/ZonePartitionReplicaListenerTest.java:
##########
@@ -1275,6 +1323,492 @@ void exceptionIsReturnedIfTableProcessorIsAbsent() {
assertThat(ex.getMessage(), is("Table is already destroyed [tableId=1]"));
}
+ @Test
+ public void testTxStateReplicaRequestEmptyState() throws Exception {
+ doAnswer(invocation -> {
+ UUID txId = invocation.getArgument(5);
+
+ txManager.updateTxMeta(txId, old -> new TxStateMeta(
+ ABORTED,
+ localNode.id(),
+ commitPartitionId,
+ null,
+ null,
+ null
+ ));
+
+ return nullCompletedFuture();
+ }).when(txManager).finish(any(), any(), anyBoolean(), anyBoolean(), any(), any());
+
+ CompletableFuture<ReplicaResult> fut = zonePartitionReplicaListener.invoke(TX_MESSAGES_FACTORY.txStateCommitPartitionRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .txId(newTxId())
+ .enlistmentConsistencyToken(ANY_ENLISTMENT_CONSISTENCY_TOKEN)
+ .build(), randomUUID());
+
+ assertThat(fut, willSucceedFast());
+
+ TransactionMeta txMeta = (TransactionMeta) fut.get().result();
+
+ assertNotNull(txMeta);
+
+ assertEquals(ABORTED, txMeta.txState());
+ }
+
+ @Test
+ public void testTxStateReplicaRequestCommitState() throws Exception {
+ UUID txId = newTxId();
+
+ txStateStorage.putForRebalance(txId, new TxMeta(COMMITTED, singletonList(new EnlistedPartitionGroup(grpId)), clock.now()));
+
+ HybridTimestamp readTimestamp = clock.now();
+
+ CompletableFuture<ReplicaResult> fut = zonePartitionReplicaListener.invoke(TX_MESSAGES_FACTORY.txStateCommitPartitionRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .txId(txId)
+ .enlistmentConsistencyToken(ANY_ENLISTMENT_CONSISTENCY_TOKEN)
+ .build(), localNode.id());
+
+ assertThat(fut, willSucceedFast());
+
+ TransactionMeta txMeta = (TransactionMeta) fut.get().result();
+
+ assertNotNull(txMeta);
+ assertEquals(COMMITTED, txMeta.txState());
+ assertNotNull(txMeta.commitTimestamp());
+ assertTrue(readTimestamp.compareTo(txMeta.commitTimestamp()) > 0);
+ }
+
+ @Test
+ @WithSystemProperty(key = COLOCATION_FEATURE_FLAG, value = "true")
+ public void testWriteIntentOnPrimaryReplicaSingleUpdate() {
+ zonePartitionReplicaListener.addTableReplicaProcessor(TABLE_ID, mocked -> tableReplicaProcessor);
+
+ UUID txId = newTxId();
+ AtomicInteger counter = new AtomicInteger();
+
+ testWriteIntentOnPrimaryReplica(
+ txId,
+ () -> {
+ BinaryRow binaryRow = binaryRow(counter.getAndIncrement());
+
+ return TABLE_MESSAGES_FACTORY.readWriteSingleRowReplicaRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .tableId(TABLE_ID)
+ .transactionId(txId)
+ .requestType(RW_INSERT)
+ .schemaVersion(binaryRow.schemaVersion())
+ .binaryTuple(binaryRow.tupleSlice())
+ .enlistmentConsistencyToken(ANY_ENLISTMENT_CONSISTENCY_TOKEN)
+ .commitPartitionId(commitPartitionId())
+ .coordinatorId(localNode.id())
+ .timestamp(clock.now())
+ .build();
+ },
+ () -> checkRowInMvStorage(binaryRow(0), true)
+ );
+
+ cleanup(txId);
+
+ zonePartitionReplicaListener.removeTableReplicaProcessor(TABLE_ID);
+ }
+
+
+ @Test
+ @WithSystemProperty(key = COLOCATION_FEATURE_FLAG, value = "true")
+ public void testWriteIntentOnPrimaryReplicaUpdateAll() {
+ zonePartitionReplicaListener.addTableReplicaProcessor(TABLE_ID, mocked -> tableReplicaProcessor);
+
+ UUID txId = newTxId();
+ AtomicInteger counter = new AtomicInteger();
+
+ testWriteIntentOnPrimaryReplica(
+ txId,
+ () -> {
+ int cntr = counter.getAndIncrement();
+ BinaryRow binaryRow0 = binaryRow(cntr * 2);
+ BinaryRow binaryRow1 = binaryRow(cntr * 2 + 1);
+
+ return TABLE_MESSAGES_FACTORY.readWriteMultiRowReplicaRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .tableId(TABLE_ID)
+ .transactionId(txId)
+ .requestType(RW_UPSERT_ALL)
+ .schemaVersion(binaryRow0.schemaVersion())
+ .binaryTuples(asList(binaryRow0.tupleSlice(), binaryRow1.tupleSlice()))
+ .enlistmentConsistencyToken(ANY_ENLISTMENT_CONSISTENCY_TOKEN)
+ .commitPartitionId(commitPartitionId())
+ .coordinatorId(localNode.id())
+ .timestamp(clock.now())
+ .build();
+ },
+ () -> checkRowInMvStorage(binaryRow(0), true)
+ );
+
+ cleanup(txId);
+
+ zonePartitionReplicaListener.removeTableReplicaProcessor(TABLE_ID);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ @WithSystemProperty(key = COLOCATION_FEATURE_FLAG, value = "true")
+ void writeIntentSwitchForCompactedCatalogTimestampWorks(boolean commit) {
+ int earliestVersion = 999;
+ Catalog mockEarliestCatalog = mock(Catalog.class);
+ when(mockEarliestCatalog.version()).thenReturn(earliestVersion);
+
+ UUID txId = newTxId();
+ HybridTimestamp beginTs = beginTimestamp(txId);
+ HybridTimestamp commitTs = clock.now();
+
+ HybridTimestamp reliableCatalogVersionTs = commit ? commitTs : beginTs;
+ when(catalogService.activeCatalog(reliableCatalogVersionTs.longValue())).thenThrow(new CatalogNotFoundException("Oops"));
+ when(catalogService.earliestCatalog()).thenReturn(mockEarliestCatalog);
+
+ zonePartitionReplicaListener.addTableReplicaProcessor(TABLE_ID, mocked -> tableReplicaProcessor);
+
+ CompletableFuture<ReplicaResult> invokeFuture = zonePartitionReplicaListener.invoke(
+ TX_MESSAGES_FACTORY.writeIntentSwitchReplicaRequest()
+ .groupId(tablePartitionIdMessage(grpId))
+ .tableIds(Set.of(grpId.tableId()))
+ .txId(txId)
+ .commit(commit)
+ .commitTimestamp(commit ? commitTs : null)
+ .build(),
+ localNode.id()
+ );
+
+ assertThat(invokeFuture, willCompleteSuccessfully());
+ assertThat(invokeFuture.join().applyResult().replicationFuture(), willCompleteSuccessfully());
+
+ verify(mockRaftClient).run(commandCaptor.capture());
+ WriteIntentSwitchCommand command = (WriteIntentSwitchCommand) commandCaptor.getValue();
+
+ assertThat(command.requiredCatalogVersion(), is(earliestVersion));
+
+ zonePartitionReplicaListener.removeTableReplicaProcessor(TABLE_ID);
+ }
+
+
+ @Test
+ public void abortsSuccessfully() {
+ AtomicReference<Boolean> committed = interceptFinishTxCommand();
+
+ CompletableFuture<?> future = beginAndAbortTx();
+
+ assertThat(future, willSucceedFast());
+
+ assertThat(committed.get(), is(false));
+ }
+
+ @Test
+ public void commitsOnSameSchemaSuccessfully() {
+ when(validationSchemasSource.tableSchemaVersionsBetween(anyInt(), any(), any(HybridTimestamp.class)))
+ .thenReturn(List.of(
+ tableSchema(CURRENT_SCHEMA_VERSION, List.of(nullableColumn("col")))
+ ));
+
+ AtomicReference<Boolean> committed = interceptFinishTxCommand();
+
+ CompletableFuture<?> future = beginAndCommitTx();
+
+ assertThat(future, willSucceedFast());
+
+ assertThat(committed.get(), is(true));
+ }
+
+
Review Comment:
And removed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-24770 Port Table Partition Listeners Test logic to Zone Partition Listener Tests [ignite-3]
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch merged PR #6004:
URL: https://github.com/apache/ignite-3/pull/6004
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-24770 Copy PartitionReplicaListenerTest logic to test ZonePartitionReplicaListener [ignite-3]
Posted by "JAkutenshi (via GitHub)" <gi...@apache.org>.
JAkutenshi commented on code in PR #6004:
URL: https://github.com/apache/ignite-3/pull/6004#discussion_r2137573643
##########
modules/partition-replicator/src/test/java/org/apache/ignite/internal/partition/replicator/raft/ZonePartitionRaftListenerTest.java:
##########
@@ -367,6 +405,151 @@ void processorsAreNotInitializedWithoutSnapshot(@Mock RaftTableProcessor tablePr
verify(tableProcessor, never()).initialize(any(), any(), anyLong(), anyLong());
}
+ @Test
+ void testSkipWriteCommandByAppliedIndex() {
+ mvPartitionStorage = spy(new TestMvPartitionStorage(PARTITION_ID));
+
+ PartitionListener tableProcessor = partitionListener(TABLE_ID);
+
+ listener.addTableProcessor(TABLE_ID, tableProcessor);
+ // Update(All)Command handling requires both information about raft group topology and the primary replica,
+ // thus onConfigurationCommited and primaryReplicaChangeCommand are called.
+ AtomicInteger raftIndex = new AtomicInteger();
+
+ long index = raftIndex.incrementAndGet();
+
+ listener.onConfigurationCommitted(
+ new RaftGroupConfiguration(
+ index,
+ 1,
+ List.of("foo"),
+ List.of("bar"),
+ null,
+ null
+ ),
+ index,
+ 1
+ );
+
+ PrimaryReplicaChangeCommand command = REPLICA_MESSAGES_FACTORY.primaryReplicaChangeCommand()
+ .primaryReplicaNodeName("primary")
+ .primaryReplicaNodeId(randomUUID())
+ .leaseStartTime(HybridTimestamp.MIN_VALUE.addPhysicalTime(1).longValue())
+ .build();
+
+ listener.onWrite(List.of(writeCommandCommandClosure(raftIndex.incrementAndGet(), 1, command, null, null)).iterator());
+
+ mvPartitionStorage.lastApplied(10L, 1L);
+
+ UpdateCommand updateCommand = mock(UpdateCommand.class);
+ when(updateCommand.tableId()).thenReturn(TABLE_ID);
+
+ WriteIntentSwitchCommand writeIntentSwitchCommand = mock(WriteIntentSwitchCommand.class);
+
+ SafeTimeSyncCommand safeTimeSyncCommand = mock(SafeTimeSyncCommand.class);
+
+ FinishTxCommand finishTxCommand = mock(FinishTxCommand.class);
+ when(finishTxCommand.groupType()).thenReturn(PartitionReplicationMessageGroup.GROUP_TYPE);
+ when(finishTxCommand.messageType()).thenReturn(Commands.FINISH_TX);
+
+ PrimaryReplicaChangeCommand primaryReplicaChangeCommand = mock(PrimaryReplicaChangeCommand.class);
+
+ // Checks for MvPartitionStorage.
+ listener.onWrite(List.of(
+ writeCommandCommandClosure(3, 1, updateCommand, updateCommandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(10, 1, updateCommand, updateCommandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(4, 1, writeIntentSwitchCommand, commandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(5, 1, safeTimeSyncCommand, commandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(6, 1, primaryReplicaChangeCommand, commandClosureResultCaptor, null)
+ ).iterator());
+
+ // Two storage runConsistently runs are expected: one for configuration application and another for primaryReplicaChangeCommand
+ // handling. Both comes from initial configuration preparation in @BeforeEach
+ verify(mvPartitionStorage, times(2)).runConsistently(any(WriteClosure.class));
+ verify(mvPartitionStorage, times(3)).lastApplied(anyLong(), anyLong()); // !! 3 vs 1
+
+ List<UpdateCommandResult> allValues = updateCommandClosureResultCaptor.getAllValues();
+ assertThat(allValues, containsInAnyOrder(new Throwable[]{null, null}));
+ assertThat(commandClosureResultCaptor.getAllValues(), containsInAnyOrder(new Throwable[]{null, null, null}));
+
+ // Checks for TxStateStorage.
+ mvPartitionStorage.lastApplied(1L, 1L);
+ txStatePartitionStorage.lastApplied(10L, 2L);
+
+ commandClosureResultCaptor = ArgumentCaptor.forClass(Throwable.class);
+
+ listener.onWrite(List.of(
+ writeCommandCommandClosure(2, 1, finishTxCommand, commandClosureResultCaptor, clock.now()),
+ writeCommandCommandClosure(10, 1, finishTxCommand, commandClosureResultCaptor, clock.now())
+ ).iterator());
+
+ verify(txStatePartitionStorage, never())
+ .compareAndSet(any(UUID.class), any(TxState.class), any(TxMeta.class), anyLong(), anyLong());
+ verify(txStatePartitionStorage, times(1)).lastApplied(anyLong(), anyLong());
+
+ assertThat(commandClosureResultCaptor.getAllValues(), containsInAnyOrder(new Throwable[]{null, null}));
+
+ listener.removeTableProcessor(TABLE_ID);
+ }
+
+
+ @Test
+ void updatesLastAppliedForFinishTxCommands() {
+ safeTimeClock.update(clock.now(), null);
+
+ FinishTxCommand command = PARTITION_REPLICATION_MESSAGES_FACTORY.finishTxCommand()
+ .txId(TestTransactionIds.newTransactionId())
+ .initiatorTime(clock.now())
+ .partitions(List.of())
+ .build();
+
+ listener.onWrite(List.of(
+ writeCommandCommandClosure(3, 2, command)
+ ).iterator());
+
+ assertThat(txStatePartitionStorage.lastAppliedIndex(), is(3L));
+ assertThat(txStatePartitionStorage.lastAppliedTerm(), is(2L));
+ }
+
+ private CommandClosure<WriteCommand> writeCommandCommandClosure(
Review Comment:
Applied.
##########
modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/ZonePartitionReplicaListenerTest.java:
##########
@@ -237,6 +277,10 @@ public class ZonePartitionReplicaListenerTest extends IgniteAbstractTest {
private final LockManager lockManager = lockManager();
+ @Captor
+ private ArgumentCaptor<Command> commandCaptor;
+
+
Review Comment:
Yes, removed.
##########
modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/ZonePartitionReplicaListenerTest.java:
##########
@@ -322,7 +366,7 @@ public class ZonePartitionReplicaListenerTest extends IgniteAbstractTest {
private TopologyService topologySrv;
@Mock
- private PendingComparableValuesTracker<HybridTimestamp, Void> safeTimeClock;
+ private SafeTimeValuesTracker safeTimeClock;
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org