rpuch commented on code in PR #2155:
URL: https://github.com/apache/ignite-3/pull/2155#discussion_r1221006155
##########
modules/storage-api/src/main/java/org/apache/ignite/internal/storage/util/StorageOperation.java:
##########
@@ -109,6 +109,34 @@ String inProcessErrorMessage(String storageInfo) {
* Storage rebalancing start operation.
*/
static class StartRebalanceStorageOperation extends StorageOperation {
+ private final AtomicReference<AbortRebalanceStorageOperation>
abortRebalanceOperation = new AtomicReference<>();
+
+ private final CompletableFuture<Void> startRebalanceFuture = new
CompletableFuture<>();
+
+ /**
+ * Attempts to set the abort rebalance operation.
+ *
+ * @param abortRebalanceOperation Abort rebalance operation.
+ * @return {@code True} if the operation was set by current method
invocation, {@code false} if by another method invocation.
+ */
+ boolean setAbortOperation(AbortRebalanceStorageOperation
abortRebalanceOperation) {
+ return this.abortRebalanceOperation.compareAndSet(null,
abortRebalanceOperation);
+ }
+
+ /**
+ * Returns {@link #setAbortOperation(AbortRebalanceStorageOperation)
set} a abort rebalance operation.
Review Comment:
```suggestion
* Returns the {@link
#setAbortOperation(AbortRebalanceStorageOperation) set} abort rebalance
operation.
```
##########
modules/storage-api/src/main/java/org/apache/ignite/internal/storage/util/MvPartitionStorages.java:
##########
@@ -415,7 +430,15 @@ private String
createStorageInProgressOfRebalanceErrorMessage(int partitionId) {
return operation;
}
- return operation instanceof DestroyStorageOperation ?
((DestroyStorageOperation) operation).getCreateStorageOperation() : null;
+ if (operation instanceof DestroyStorageOperation) {
+ return ((DestroyStorageOperation)
operation).getCreateStorageOperation();
+ }
+
+ if (operation instanceof StartRebalanceStorageOperation) {
+ return ((StartRebalanceStorageOperation)
operation).getAbortRebalanceOperation();
+ }
+
+ return null;
Review Comment:
It seems that `completeOperation()` does not describe very accurately what
this method does. It's more about choosing the 'next' operation. How about
renaming it? Maybe, to `nextOperationIfAvailable()`?
##########
modules/storage-api/src/main/java/org/apache/ignite/internal/storage/util/MvPartitionStorages.java:
##########
@@ -238,14 +240,16 @@ public CompletableFuture<Void> startRebalance(int
partitionId, Function<T, Compl
assert old == null : createStorageInfo(partitionId);
return startRebalanceFuture;
- }).whenComplete((unused, throwable) ->
- operationByPartitionId.compute(partitionId, (partId,
operation) -> {
- assert operation instanceof
StartRebalanceStorageOperation :
- createStorageInfo(partitionId) + ", op=" +
operation;
+ }).whenComplete((unused, throwable) -> {
+ operationByPartitionId.compute(partitionId, (partId,
operation) -> {
+ assert operation instanceof
StartRebalanceStorageOperation : createStorageInfo(partitionId) + ", op=" +
operation;
- return completeOperation(operation);
- })
- );
+ return completeOperation(operation);
+ });
+
+ // Even if an error occurs, we must be able to abort the
rebalance, so we do not report the error.
Review Comment:
Does this mean that we swallow the error?
##########
modules/storage-api/src/main/java/org/apache/ignite/internal/storage/util/StorageOperation.java:
##########
@@ -109,6 +109,34 @@ String inProcessErrorMessage(String storageInfo) {
* Storage rebalancing start operation.
*/
static class StartRebalanceStorageOperation extends StorageOperation {
+ private final AtomicReference<AbortRebalanceStorageOperation>
abortRebalanceOperation = new AtomicReference<>();
Review Comment:
It looks like a comment explaining why we need this (an abortion waiting for
the start to ... ahem... finish, probably) could be useful
##########
modules/storage-api/src/main/java/org/apache/ignite/internal/storage/util/StorageOperation.java:
##########
@@ -109,6 +109,34 @@ String inProcessErrorMessage(String storageInfo) {
* Storage rebalancing start operation.
*/
static class StartRebalanceStorageOperation extends StorageOperation {
+ private final AtomicReference<AbortRebalanceStorageOperation>
abortRebalanceOperation = new AtomicReference<>();
+
+ private final CompletableFuture<Void> startRebalanceFuture = new
CompletableFuture<>();
+
+ /**
+ * Attempts to set the abort rebalance operation.
+ *
+ * @param abortRebalanceOperation Abort rebalance operation.
+ * @return {@code True} if the operation was set by current method
invocation, {@code false} if by another method invocation.
Review Comment:
```suggestion
* @return {@code true} if the operation was set by the current
method invocation, {@code false} if by another method invocation.
```
##########
modules/storage-api/src/test/java/org/apache/ignite/internal/storage/util/MvPartitionStoragesTest.java:
##########
@@ -326,8 +326,6 @@ void testStartRebalance() {
assertThrowsWithMessage(StorageRebalanceException.class, () ->
startRebalanceMvStorage(0),
"Storage in the process of starting a rebalance");
- assertThrowsWithMessage(StorageRebalanceException.class, () ->
abortRebalanceMvStorage(0),
- "Storage in the process of starting a rebalance");
Review Comment:
Should we instead assert that it does NOT throw such an exception?
--
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]
