You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2021/12/21 01:33:53 UTC
[pulsar] 17/18: fix fix-13004 Race condition in ResourceLockImpl#revalidate (#13006)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit db1a34cfe708c182b9fa54515e0a26c3c042163b
Author: JiangHaiting <ji...@apache.org>
AuthorDate: Tue Nov 30 01:32:22 2021 +0800
fix fix-13004 Race condition in ResourceLockImpl#revalidate (#13006)
(cherry picked from commit bb2c9345bc42d61090897757a9f7435c138c8ee3)
---
.../coordination/impl/ResourceLockImpl.java | 28 ++++++++++++++++++++++
1 file changed, 28 insertions(+)
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java
index 677ace7..ec9d3c4 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java
@@ -44,6 +44,7 @@ public class ResourceLockImpl<T> implements ResourceLock<T> {
private long version;
private final CompletableFuture<Void> expiredFuture;
private boolean revalidateAfterReconnection = false;
+ private CompletableFuture<Void> revalidateFuture;
private enum State {
Init,
@@ -144,6 +145,9 @@ public class ResourceLockImpl<T> implements ResourceLock<T> {
// Simple operation of acquiring the lock with no retries, or checking for the lock content
private CompletableFuture<Void> acquireWithNoRevalidation(T newValue) {
+ if (log.isDebugEnabled()) {
+ log.debug("acquireWithNoRevalidation,newValue={},version={}", newValue, version);
+ }
byte[] payload;
try {
payload = serde.serialize(path, newValue);
@@ -212,6 +216,30 @@ public class ResourceLockImpl<T> implements ResourceLock<T> {
}
synchronized CompletableFuture<Void> revalidate(T newValue) {
+ if (revalidateFuture == null || revalidateFuture.isDone()) {
+ revalidateFuture = doRevalidate(newValue);
+ } else {
+ if (log.isDebugEnabled()) {
+ log.debug("Previous revalidating is not finished while revalidate newValue={}, value={}, version={}",
+ newValue, value, version);
+ }
+ CompletableFuture<Void> newFuture = new CompletableFuture<>();
+ revalidateFuture.whenComplete((unused, throwable) -> {
+ doRevalidate(newValue).thenRun(() -> newFuture.complete(null))
+ .exceptionally(throwable1 -> {
+ newFuture.completeExceptionally(throwable1);
+ return null;
+ });
+ });
+ revalidateFuture = newFuture;
+ }
+ return revalidateFuture;
+ }
+
+ private synchronized CompletableFuture<Void> doRevalidate(T newValue) {
+ if (log.isDebugEnabled()) {
+ log.debug("doRevalidate with newValue={}, version={}", newValue, version);
+ }
return store.get(path)
.thenCompose(optGetResult -> {
if (!optGetResult.isPresent()) {