Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Retry map and cache partition destroy operations
When the partition is migrating, the operations might fail with a PartitionMigratingException. We then need to retry the operation as otherwise we are leaking memory. As for other partition operations, we wait for the migration to complete, using the default try count and wait count. Fixes: https://github.com/hazelcast/hazelcast-enterprise/issues/930 Fixes: https://github.com/hazelcast/hazelcast-enterprise/issues/1933
- Loading branch information
Matko Medenjak
committed
Mar 27, 2018
1 parent
37330c5
commit 890c299
Showing
5 changed files
with
290 additions
and
22 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
118 changes: 118 additions & 0 deletions
118
hazelcast/src/main/java/com/hazelcast/internal/util/LocalRetryableExecution.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,118 @@ | ||
| /* | ||
| * Copyright (c) 2008-2018, Hazelcast, Inc. All Rights Reserved. | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package com.hazelcast.internal.util; | ||
|
|
||
| import com.hazelcast.logging.ILogger; | ||
| import com.hazelcast.spi.NodeEngine; | ||
| import com.hazelcast.spi.Operation; | ||
| import com.hazelcast.spi.OperationResponseHandler; | ||
| import com.hazelcast.spi.exception.RetryableHazelcastException; | ||
| import com.hazelcast.spi.properties.GroupProperty; | ||
|
|
||
| import java.util.concurrent.CountDownLatch; | ||
| import java.util.concurrent.TimeUnit; | ||
| import java.util.logging.Level; | ||
|
|
||
| import static java.util.logging.Level.FINEST; | ||
| import static java.util.logging.Level.WARNING; | ||
|
|
||
| /** | ||
| * Class encapsulating local execution with retry logic. The operation must | ||
| * not have an {@link OperationResponseHandler} set and it must return | ||
| * response. | ||
| * The retry will use the configured | ||
| * {@link GroupProperty#INVOCATION_MAX_RETRY_COUNT} and | ||
| * {@link GroupProperty#INVOCATION_RETRY_PAUSE}. | ||
| * | ||
| * @see Operation#returnsResponse() | ||
| * @see Operation#getOperationResponseHandler() | ||
| * @see GroupProperty#INVOCATION_MAX_RETRY_COUNT | ||
| * @see GroupProperty#INVOCATION_RETRY_PAUSE | ||
| * @see InvocationUtil#executeLocallyWithRetry(NodeEngine, Operation) | ||
| */ | ||
| public class LocalRetryableExecution implements Runnable, OperationResponseHandler { | ||
| /** Number of times an operation is retried before being logged at WARNING level */ | ||
| private static final int LOG_MAX_INVOCATION_COUNT = 99; | ||
| private final ILogger logger; | ||
| private final CountDownLatch done = new CountDownLatch(1); | ||
| private final Operation op; | ||
| private final NodeEngine nodeEngine; | ||
| private final long invocationRetryPauseMillis; | ||
| private final int invocationMaxRetryCount; | ||
| private volatile Object response; | ||
| private int tryCount; | ||
|
|
||
| LocalRetryableExecution(NodeEngine nodeEngine, Operation op) { | ||
| this.nodeEngine = nodeEngine; | ||
| this.logger = nodeEngine.getLogger(LocalRetryableExecution.class); | ||
| this.invocationMaxRetryCount = nodeEngine.getProperties().getInteger(GroupProperty.INVOCATION_MAX_RETRY_COUNT); | ||
| this.invocationRetryPauseMillis = nodeEngine.getProperties().getMillis(GroupProperty.INVOCATION_RETRY_PAUSE); | ||
| this.op = op; | ||
| op.setOperationResponseHandler(this); | ||
| } | ||
|
|
||
| /** | ||
| * Causes the current thread to wait until the operation has finished the | ||
| * thread is {@linkplain Thread#interrupt interrupted}, or the specified | ||
| * waiting time elapses. The operation might have finished because it has | ||
| * completed (successfully or with an error) or the retry count has been | ||
| * exceeded. | ||
| * | ||
| * @param timeout the maximum time to wait | ||
| * @param unit the time unit of the {@code timeout} argument | ||
| * @return {@code true} if the operation completed or the operation retry | ||
| * count has been exceeded, else {@code false} | ||
| * @throws InterruptedException if the current thread is interrupted | ||
| * while waiting | ||
| */ | ||
| public boolean awaitCompletion(long timeout, TimeUnit unit) throws InterruptedException { | ||
| return done.await(timeout, unit); | ||
| } | ||
|
|
||
| /** | ||
| * The response of the operation execution. It may be an exception if the | ||
| * exception was not an instance of {@link RetryableHazelcastException} or | ||
| * the maximum number of retry counts was exceeded. | ||
| * The response may be also {@code null} if the operation has no response | ||
| * or the operation has not completed yet. | ||
| * | ||
| * @return the operation response | ||
| */ | ||
| public Object getResponse() { | ||
| return response; | ||
| } | ||
|
|
||
| @Override | ||
| public void run() { | ||
| nodeEngine.getOperationService().execute(op); | ||
| } | ||
|
|
||
| @Override | ||
| public void sendResponse(Operation op, Object response) { | ||
| tryCount++; | ||
| if (response instanceof RetryableHazelcastException && tryCount < invocationMaxRetryCount) { | ||
| Level level = tryCount > LOG_MAX_INVOCATION_COUNT ? WARNING : FINEST; | ||
| if (logger.isLoggable(level)) { | ||
| logger.log(level, "Retrying local execution: " + toString() + ", Reason: " + response); | ||
| } | ||
| nodeEngine.getExecutionService().schedule(this, invocationRetryPauseMillis, TimeUnit.MILLISECONDS); | ||
| } else { | ||
| this.response = response; | ||
| done.countDown(); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.