rpuch commented on code in PR #1615:
URL: https://github.com/apache/ignite-3/pull/1615#discussion_r1094400527
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItRebalanceDistributedTest.java:
##########
@@ -642,20 +815,156 @@ void stop() throws Exception {
NetworkAddress address() {
return clusterService.topologyService().localMember().address();
}
+
+ @Nullable TablePartitionId getTablePartitionId(WatchEvent event) {
+ assertTrue(event.single(), event.toString());
+
+ Entry stableAssignmentsWatchEvent = event.entryEvent().newEntry();
+
+ if (stableAssignmentsWatchEvent.value() == null) {
+ return null;
+ }
+
+ int partitionId =
extractPartitionNumber(stableAssignmentsWatchEvent.key());
+ UUID tableId = extractTableId(stableAssignmentsWatchEvent.key(),
STABLE_ASSIGNMENTS_PREFIX);
+
+ return new TablePartitionId(tableId, partitionId);
+ }
+
+ TablePartitionId getTablePartitionId(String tableName, int
partitionId) {
+ InternalTable internalTable = getInternalTable(this, tableName);
+
+ return new TablePartitionId(internalTable.tableId(), partitionId);
+ }
}
/**
* Starts the Vault component.
*/
private static VaultManager createVault(Path workDir) {
- Path vaultPath = workDir.resolve(Paths.get("vault"));
+ return new VaultManager(new PersistentVaultService(resolveDir(workDir,
"vault")));
+ }
+
+ private static Path resolveDir(Path workDir, String dirName) {
+ Path newDirPath = workDir.resolve(dirName);
try {
- Files.createDirectories(vaultPath);
+ return Files.createDirectories(newDirPath);
} catch (IOException e) {
throw new IgniteInternalException(e);
}
+ }
+
+ private static TableDefinition createTableDefinition(String tableName) {
+ return SchemaBuilders.tableBuilder("PUBLIC", tableName).columns(
+ SchemaBuilders.column("key", ColumnType.INT64).build(),
+ SchemaBuilders.column("val",
ColumnType.INT32).asNullable(true).build()
+ ).withPrimaryKey("key").build();
+ }
+
+ private void createTableForOnePartition(String tableName, int replicas,
boolean testDataStorage) {
Review Comment:
```suggestion
private void createTableWithOnePartition(String tableName, int replicas,
boolean testDataStorage) {
```
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItRebalanceDistributedTest.java:
##########
@@ -642,20 +815,156 @@ void stop() throws Exception {
NetworkAddress address() {
return clusterService.topologyService().localMember().address();
}
+
+ @Nullable TablePartitionId getTablePartitionId(WatchEvent event) {
+ assertTrue(event.single(), event.toString());
+
+ Entry stableAssignmentsWatchEvent = event.entryEvent().newEntry();
+
+ if (stableAssignmentsWatchEvent.value() == null) {
+ return null;
+ }
+
+ int partitionId =
extractPartitionNumber(stableAssignmentsWatchEvent.key());
+ UUID tableId = extractTableId(stableAssignmentsWatchEvent.key(),
STABLE_ASSIGNMENTS_PREFIX);
+
+ return new TablePartitionId(tableId, partitionId);
+ }
+
+ TablePartitionId getTablePartitionId(String tableName, int
partitionId) {
+ InternalTable internalTable = getInternalTable(this, tableName);
+
+ return new TablePartitionId(internalTable.tableId(), partitionId);
+ }
}
/**
* Starts the Vault component.
*/
private static VaultManager createVault(Path workDir) {
- Path vaultPath = workDir.resolve(Paths.get("vault"));
+ return new VaultManager(new PersistentVaultService(resolveDir(workDir,
"vault")));
+ }
+
+ private static Path resolveDir(Path workDir, String dirName) {
+ Path newDirPath = workDir.resolve(dirName);
try {
- Files.createDirectories(vaultPath);
+ return Files.createDirectories(newDirPath);
} catch (IOException e) {
throw new IgniteInternalException(e);
}
+ }
+
+ private static TableDefinition createTableDefinition(String tableName) {
+ return SchemaBuilders.tableBuilder("PUBLIC", tableName).columns(
+ SchemaBuilders.column("key", ColumnType.INT64).build(),
+ SchemaBuilders.column("val",
ColumnType.INT32).asNullable(true).build()
+ ).withPrimaryKey("key").build();
+ }
+
+ private void createTableForOnePartition(String tableName, int replicas,
boolean testDataStorage) {
+ assertThat(
+ nodes.get(0).tableManager.createTableAsync(
+ tableName,
+ tableChange -> {
+
SchemaConfigurationConverter.convert(createTableDefinition(tableName),
tableChange)
+ .changeReplicas(replicas)
+ .changePartitions(1);
+
+ if (testDataStorage) {
+
tableChange.changeDataStorage(dataStorageChange ->
dataStorageChange.convert(TestDataStorageChange.class));
+ }
+ }
+ ),
+ willCompleteSuccessfully()
+ );
+
+ assertEquals(replicas, getPartitionClusterNodes(0, 0).size());
+ assertEquals(replicas, getPartitionClusterNodes(1, 0).size());
+ assertEquals(replicas, getPartitionClusterNodes(2, 0).size());
+ }
+
+ private void changeTableReplicasForSinglePartition(String tableName, int
replicas) {
+ assertThat(
+ nodes.get(0).tableManager.alterTableAsync(tableName,
tableChange -> {
+ tableChange.changeReplicas(replicas);
+
+ return true;
+ }),
+ willCompleteSuccessfully()
+ );
+
+ waitPartitionAssignmentsSyncedToExpected(0, replicas);
+
+ assertEquals(replicas, getPartitionClusterNodes(0, 0).size());
+ assertEquals(replicas, getPartitionClusterNodes(1, 0).size());
+ assertEquals(replicas, getPartitionClusterNodes(2, 0).size());
+ }
+
+ private static Set<Assignment> getEvictedAssignments(Set<Assignment>
beforeChange, Set<Assignment> afterChange) {
+ Set<Assignment> result = new HashSet<>(beforeChange);
+
+ result.removeAll(afterChange);
+
+ return result;
+ }
+
+ private static @Nullable InternalTable getInternalTable(Node node, String
tableName) {
+ Table table = node.tableManager.table(tableName);
+
+ assertNotNull(table, tableName);
+
+ return ((TableImpl) table).internalTable();
+ }
+
+ private static void checkInvokeDestroyPartitionStorages(Node node, String
tableName, int partitionId) {
+ InternalTable internalTable = getInternalTable(node, tableName);
+
+ verify(internalTable.storage(),
times(1)).destroyPartition(eq(partitionId));
+ verify(internalTable.txStateStorage(),
times(1)).destroyTxStateStorage(eq(partitionId));
+ }
+
+ private static void throwExceptionOnInvokeDestroyPartitionStorages(Node
node, String tableName, int partitionId) {
+ InternalTable internalTable = getInternalTable(node, tableName);
+
+ doAnswer(answer -> CompletableFuture.failedFuture(new
StorageException("From test")))
+ .when(internalTable.storage())
+ .destroyPartition(eq(partitionId));
Review Comment:
```suggestion
.destroyPartition(partitionId);
```
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItRebalanceDistributedTest.java:
##########
@@ -642,20 +815,156 @@ void stop() throws Exception {
NetworkAddress address() {
return clusterService.topologyService().localMember().address();
}
+
+ @Nullable TablePartitionId getTablePartitionId(WatchEvent event) {
+ assertTrue(event.single(), event.toString());
+
+ Entry stableAssignmentsWatchEvent = event.entryEvent().newEntry();
+
+ if (stableAssignmentsWatchEvent.value() == null) {
+ return null;
+ }
+
+ int partitionId =
extractPartitionNumber(stableAssignmentsWatchEvent.key());
+ UUID tableId = extractTableId(stableAssignmentsWatchEvent.key(),
STABLE_ASSIGNMENTS_PREFIX);
+
+ return new TablePartitionId(tableId, partitionId);
+ }
+
+ TablePartitionId getTablePartitionId(String tableName, int
partitionId) {
+ InternalTable internalTable = getInternalTable(this, tableName);
+
+ return new TablePartitionId(internalTable.tableId(), partitionId);
+ }
}
/**
* Starts the Vault component.
*/
private static VaultManager createVault(Path workDir) {
- Path vaultPath = workDir.resolve(Paths.get("vault"));
+ return new VaultManager(new PersistentVaultService(resolveDir(workDir,
"vault")));
+ }
+
+ private static Path resolveDir(Path workDir, String dirName) {
+ Path newDirPath = workDir.resolve(dirName);
try {
- Files.createDirectories(vaultPath);
+ return Files.createDirectories(newDirPath);
} catch (IOException e) {
throw new IgniteInternalException(e);
}
+ }
+
+ private static TableDefinition createTableDefinition(String tableName) {
+ return SchemaBuilders.tableBuilder("PUBLIC", tableName).columns(
+ SchemaBuilders.column("key", ColumnType.INT64).build(),
+ SchemaBuilders.column("val",
ColumnType.INT32).asNullable(true).build()
+ ).withPrimaryKey("key").build();
+ }
+
+ private void createTableForOnePartition(String tableName, int replicas,
boolean testDataStorage) {
+ assertThat(
+ nodes.get(0).tableManager.createTableAsync(
+ tableName,
+ tableChange -> {
+
SchemaConfigurationConverter.convert(createTableDefinition(tableName),
tableChange)
+ .changeReplicas(replicas)
+ .changePartitions(1);
+
+ if (testDataStorage) {
+
tableChange.changeDataStorage(dataStorageChange ->
dataStorageChange.convert(TestDataStorageChange.class));
+ }
+ }
+ ),
+ willCompleteSuccessfully()
+ );
+
+ assertEquals(replicas, getPartitionClusterNodes(0, 0).size());
+ assertEquals(replicas, getPartitionClusterNodes(1, 0).size());
+ assertEquals(replicas, getPartitionClusterNodes(2, 0).size());
+ }
+
+ private void changeTableReplicasForSinglePartition(String tableName, int
replicas) {
+ assertThat(
+ nodes.get(0).tableManager.alterTableAsync(tableName,
tableChange -> {
+ tableChange.changeReplicas(replicas);
+
+ return true;
+ }),
+ willCompleteSuccessfully()
+ );
+
+ waitPartitionAssignmentsSyncedToExpected(0, replicas);
+
+ assertEquals(replicas, getPartitionClusterNodes(0, 0).size());
+ assertEquals(replicas, getPartitionClusterNodes(1, 0).size());
+ assertEquals(replicas, getPartitionClusterNodes(2, 0).size());
+ }
+
+ private static Set<Assignment> getEvictedAssignments(Set<Assignment>
beforeChange, Set<Assignment> afterChange) {
+ Set<Assignment> result = new HashSet<>(beforeChange);
+
+ result.removeAll(afterChange);
+
+ return result;
+ }
+
+ private static @Nullable InternalTable getInternalTable(Node node, String
tableName) {
+ Table table = node.tableManager.table(tableName);
+
+ assertNotNull(table, tableName);
+
+ return ((TableImpl) table).internalTable();
+ }
+
+ private static void checkInvokeDestroyPartitionStorages(Node node, String
tableName, int partitionId) {
+ InternalTable internalTable = getInternalTable(node, tableName);
+
+ verify(internalTable.storage(),
times(1)).destroyPartition(eq(partitionId));
+ verify(internalTable.txStateStorage(),
times(1)).destroyTxStateStorage(eq(partitionId));
+ }
+
+ private static void throwExceptionOnInvokeDestroyPartitionStorages(Node
node, String tableName, int partitionId) {
+ InternalTable internalTable = getInternalTable(node, tableName);
+
+ doAnswer(answer -> CompletableFuture.failedFuture(new
StorageException("From test")))
+ .when(internalTable.storage())
+ .destroyPartition(eq(partitionId));
+
+ doAnswer(answer -> CompletableFuture.failedFuture(new
IgniteInternalException("From test")))
+ .when(internalTable.txStateStorage())
+ .destroyTxStateStorage(eq(partitionId));
Review Comment:
```suggestion
.destroyTxStateStorage(partitionId);
```
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java:
##########
@@ -2102,4 +2055,60 @@ private CompletableFuture<PartitionStorages>
getOrCreatePartitionStorages(TableI
}, ioExecutor)
.thenCompose(Function.identity());
}
+
+ /**
+ * Handles the {@link RebalanceUtil#STABLE_ASSIGNMENTS_PREFIX} update
event.
+ *
+ * @param evt Event.
+ */
+ protected void handleChangeStableAssignmentEvent(WatchEvent evt) {
+ inBusyLock(busyLock, () -> {
+ assert evt.single() : evt;
+
+ Entry stableAssignmentsWatchEvent = evt.entryEvent().newEntry();
+
+ if (stableAssignmentsWatchEvent.value() == null) {
+ return;
+ }
+
+ int partitionId =
extractPartitionNumber(stableAssignmentsWatchEvent.key());
+ UUID tableId = extractTableId(stableAssignmentsWatchEvent.key(),
STABLE_ASSIGNMENTS_PREFIX);
+
+ TablePartitionId tablePartitionId = new TablePartitionId(tableId,
partitionId);
+
+ Set<Assignment> stableAssignments =
ByteUtils.fromBytes(stableAssignmentsWatchEvent.value());
+
+ byte[] pendingAssignmentsFromMetaStorage = metaStorageMgr.get(
+ pendingPartAssignmentsKey(tablePartitionId),
+ stableAssignmentsWatchEvent.revision()
+ ).join().value();
+
+ Set<Assignment> pendingAssignments =
pendingAssignmentsFromMetaStorage == null
+ ? Set.of()
+ : ByteUtils.fromBytes(pendingAssignmentsFromMetaStorage);
+
+ String localMemberName =
clusterService.topologyService().localMember().name();
+
+ boolean shouldStopLocalServices =
Stream.concat(stableAssignments.stream(), pendingAssignments.stream())
+ .noneMatch(assignment ->
assignment.consistentId().equals(localMemberName));
+
+ if (shouldStopLocalServices) {
+ try {
+ raftMgr.stopRaftNodes(tablePartitionId);
+
+ replicaMgr.stopReplica(tablePartitionId);
+ } catch (NodeStoppingException e) {
+ // no-op
+ }
+
+ InternalTable internalTable =
tablesByIdVv.latest().get(tableId).internalTable();
+
+ // Should be done fairly quickly.
+ allOf(
+ internalTable.storage().destroyPartition(partitionId),
+ runAsync(() ->
internalTable.txStateStorage().destroyTxStateStorage(partitionId), ioExecutor)
+ ).join();
Review Comment:
It looks like we only destroy partition data, but the raft group data (log
and metastorage) remain untouched. The Jira issue mentions that we need to
destroy the log as well.
I think we can move the log and RAFT metastorage destruction to another
issue, but we need a TODO mentioning it.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItRebalanceDistributedTest.java:
##########
@@ -642,20 +815,156 @@ void stop() throws Exception {
NetworkAddress address() {
return clusterService.topologyService().localMember().address();
}
+
+ @Nullable TablePartitionId getTablePartitionId(WatchEvent event) {
+ assertTrue(event.single(), event.toString());
+
+ Entry stableAssignmentsWatchEvent = event.entryEvent().newEntry();
+
+ if (stableAssignmentsWatchEvent.value() == null) {
+ return null;
+ }
+
+ int partitionId =
extractPartitionNumber(stableAssignmentsWatchEvent.key());
+ UUID tableId = extractTableId(stableAssignmentsWatchEvent.key(),
STABLE_ASSIGNMENTS_PREFIX);
+
+ return new TablePartitionId(tableId, partitionId);
+ }
+
+ TablePartitionId getTablePartitionId(String tableName, int
partitionId) {
+ InternalTable internalTable = getInternalTable(this, tableName);
+
+ return new TablePartitionId(internalTable.tableId(), partitionId);
+ }
}
/**
* Starts the Vault component.
*/
private static VaultManager createVault(Path workDir) {
- Path vaultPath = workDir.resolve(Paths.get("vault"));
+ return new VaultManager(new PersistentVaultService(resolveDir(workDir,
"vault")));
+ }
+
+ private static Path resolveDir(Path workDir, String dirName) {
+ Path newDirPath = workDir.resolve(dirName);
try {
- Files.createDirectories(vaultPath);
+ return Files.createDirectories(newDirPath);
} catch (IOException e) {
throw new IgniteInternalException(e);
}
+ }
+
+ private static TableDefinition createTableDefinition(String tableName) {
+ return SchemaBuilders.tableBuilder("PUBLIC", tableName).columns(
+ SchemaBuilders.column("key", ColumnType.INT64).build(),
+ SchemaBuilders.column("val",
ColumnType.INT32).asNullable(true).build()
+ ).withPrimaryKey("key").build();
+ }
+
+ private void createTableForOnePartition(String tableName, int replicas,
boolean testDataStorage) {
+ assertThat(
+ nodes.get(0).tableManager.createTableAsync(
+ tableName,
+ tableChange -> {
+
SchemaConfigurationConverter.convert(createTableDefinition(tableName),
tableChange)
+ .changeReplicas(replicas)
+ .changePartitions(1);
+
+ if (testDataStorage) {
+
tableChange.changeDataStorage(dataStorageChange ->
dataStorageChange.convert(TestDataStorageChange.class));
+ }
+ }
+ ),
+ willCompleteSuccessfully()
+ );
+
+ assertEquals(replicas, getPartitionClusterNodes(0, 0).size());
+ assertEquals(replicas, getPartitionClusterNodes(1, 0).size());
+ assertEquals(replicas, getPartitionClusterNodes(2, 0).size());
+ }
+
+ private void changeTableReplicasForSinglePartition(String tableName, int
replicas) {
+ assertThat(
+ nodes.get(0).tableManager.alterTableAsync(tableName,
tableChange -> {
+ tableChange.changeReplicas(replicas);
+
+ return true;
+ }),
+ willCompleteSuccessfully()
+ );
+
+ waitPartitionAssignmentsSyncedToExpected(0, replicas);
+
+ assertEquals(replicas, getPartitionClusterNodes(0, 0).size());
+ assertEquals(replicas, getPartitionClusterNodes(1, 0).size());
+ assertEquals(replicas, getPartitionClusterNodes(2, 0).size());
+ }
+
+ private static Set<Assignment> getEvictedAssignments(Set<Assignment>
beforeChange, Set<Assignment> afterChange) {
+ Set<Assignment> result = new HashSet<>(beforeChange);
+
+ result.removeAll(afterChange);
+
+ return result;
+ }
+
+ private static @Nullable InternalTable getInternalTable(Node node, String
tableName) {
+ Table table = node.tableManager.table(tableName);
+
+ assertNotNull(table, tableName);
+
+ return ((TableImpl) table).internalTable();
+ }
+
+ private static void checkInvokeDestroyPartitionStorages(Node node, String
tableName, int partitionId) {
+ InternalTable internalTable = getInternalTable(node, tableName);
+
+ verify(internalTable.storage(),
times(1)).destroyPartition(eq(partitionId));
+ verify(internalTable.txStateStorage(),
times(1)).destroyTxStateStorage(eq(partitionId));
Review Comment:
```suggestion
verify(internalTable.storage(),
times(1)).destroyPartition(partitionId);
verify(internalTable.txStateStorage(),
times(1)).destroyTxStateStorage(partitionId);
```
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItRebalanceDistributedTest.java:
##########
@@ -391,6 +435,85 @@ void testRebalanceRetryWhenCatchupFailed() throws
Exception {
assertEquals(3, getPartitionClusterNodes(2, 0).size());
}
+ @Test
+ @Tag(TEST_TX_STORAGE_TAG)
Review Comment:
>Tags are used to filter which tests are executed for a given test plan
But here, they are used to signal that some customizations need to be
applied in the `before()` method. This looks like an abuse. It seems that a
custom annotation should be added instead.
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/configuration/storage/ItRebalanceDistributedTest.java:
##########
@@ -642,20 +815,156 @@ void stop() throws Exception {
NetworkAddress address() {
return clusterService.topologyService().localMember().address();
}
+
+ @Nullable TablePartitionId getTablePartitionId(WatchEvent event) {
+ assertTrue(event.single(), event.toString());
+
+ Entry stableAssignmentsWatchEvent = event.entryEvent().newEntry();
+
+ if (stableAssignmentsWatchEvent.value() == null) {
+ return null;
+ }
+
+ int partitionId =
extractPartitionNumber(stableAssignmentsWatchEvent.key());
+ UUID tableId = extractTableId(stableAssignmentsWatchEvent.key(),
STABLE_ASSIGNMENTS_PREFIX);
+
+ return new TablePartitionId(tableId, partitionId);
+ }
+
+ TablePartitionId getTablePartitionId(String tableName, int
partitionId) {
+ InternalTable internalTable = getInternalTable(this, tableName);
+
+ return new TablePartitionId(internalTable.tableId(), partitionId);
+ }
}
/**
* Starts the Vault component.
*/
private static VaultManager createVault(Path workDir) {
- Path vaultPath = workDir.resolve(Paths.get("vault"));
+ return new VaultManager(new PersistentVaultService(resolveDir(workDir,
"vault")));
+ }
+
+ private static Path resolveDir(Path workDir, String dirName) {
+ Path newDirPath = workDir.resolve(dirName);
try {
- Files.createDirectories(vaultPath);
+ return Files.createDirectories(newDirPath);
} catch (IOException e) {
throw new IgniteInternalException(e);
}
+ }
+
+ private static TableDefinition createTableDefinition(String tableName) {
+ return SchemaBuilders.tableBuilder("PUBLIC", tableName).columns(
+ SchemaBuilders.column("key", ColumnType.INT64).build(),
+ SchemaBuilders.column("val",
ColumnType.INT32).asNullable(true).build()
+ ).withPrimaryKey("key").build();
+ }
+
+ private void createTableForOnePartition(String tableName, int replicas,
boolean testDataStorage) {
+ assertThat(
+ nodes.get(0).tableManager.createTableAsync(
+ tableName,
+ tableChange -> {
+
SchemaConfigurationConverter.convert(createTableDefinition(tableName),
tableChange)
+ .changeReplicas(replicas)
+ .changePartitions(1);
+
+ if (testDataStorage) {
+
tableChange.changeDataStorage(dataStorageChange ->
dataStorageChange.convert(TestDataStorageChange.class));
+ }
+ }
+ ),
+ willCompleteSuccessfully()
+ );
+
+ assertEquals(replicas, getPartitionClusterNodes(0, 0).size());
+ assertEquals(replicas, getPartitionClusterNodes(1, 0).size());
+ assertEquals(replicas, getPartitionClusterNodes(2, 0).size());
+ }
+
+ private void changeTableReplicasForSinglePartition(String tableName, int
replicas) {
+ assertThat(
+ nodes.get(0).tableManager.alterTableAsync(tableName,
tableChange -> {
+ tableChange.changeReplicas(replicas);
+
+ return true;
+ }),
+ willCompleteSuccessfully()
+ );
+
+ waitPartitionAssignmentsSyncedToExpected(0, replicas);
+
+ assertEquals(replicas, getPartitionClusterNodes(0, 0).size());
+ assertEquals(replicas, getPartitionClusterNodes(1, 0).size());
+ assertEquals(replicas, getPartitionClusterNodes(2, 0).size());
+ }
+
+ private static Set<Assignment> getEvictedAssignments(Set<Assignment>
beforeChange, Set<Assignment> afterChange) {
+ Set<Assignment> result = new HashSet<>(beforeChange);
+
+ result.removeAll(afterChange);
+
+ return result;
+ }
+
+ private static @Nullable InternalTable getInternalTable(Node node, String
tableName) {
+ Table table = node.tableManager.table(tableName);
+
+ assertNotNull(table, tableName);
+
+ return ((TableImpl) table).internalTable();
+ }
+
+ private static void checkInvokeDestroyPartitionStorages(Node node, String
tableName, int partitionId) {
Review Comment:
```suggestion
private static void checkInvokeDestroyedPartitionStorages(Node node,
String tableName, int partitionId) {
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
