Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix: LockManagerTest.updateValue is flaky #13911

Merged
merged 2 commits into from
Jan 25, 2022
Merged

Fix: LockManagerTest.updateValue is flaky #13911

Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
bf98950
fix: LockManagerTest.updateValue is flaky
shibd Jan 23, 2022
8f828d2
code format
shibd Jan 23, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Next Next commit
fix: LockManagerTest.updateValue is flaky
  • Loading branch information
@shibd
shibd committed Jan 24, 2022
commit bf989501e4f5ef7281f477a932da104dca2d2bfb
25 changes: 15 additions & 10 deletions pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/LocalMemoryMetadataStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -168,21 +168,23 @@ public CompletableFuture<Stat> storePut(String path, byte[] data, Optional<Long>

long now = System.currentTimeMillis();

CompletableFuture<Stat> future = new CompletableFuture<>();
if (hasVersion && expectedVersion == -1) {
Value newValue = new Value(0, data, now, now, options.contains(CreateOption.Ephemeral));
Value existingValue = map.putIfAbsent(path, newValue);
if (existingValue != null) {
return FutureUtils.exception(new BadVersionException(""));
execute(() -> future.completeExceptionally(new BadVersionException("")), future);
} else {
receivedNotification(new Notification(NotificationType.Created, path));
notifyParentChildrenChanged(path);
return FutureUtils.value(new Stat(path, 0, now, now, newValue.isEphemeral(), true));
String finalPath = path;
execute(() -> future.complete(new Stat(finalPath, 0, now, now, newValue.isEphemeral(), true)), future);
}
} else {
Value existingValue = map.get(path);
long existingVersion = existingValue != null ? existingValue.version : -1;
if (hasVersion && expectedVersion != existingVersion) {
return FutureUtils.exception(new BadVersionException(""));
execute(() -> future.completeExceptionally(new BadVersionException("")), future);
} else {
long newVersion = existingValue != null ? existingValue.version + 1 : 0;
long createdTimestamp = existingValue != null ? existingValue.createdTimestamp : now;
Expand All @@ -196,12 +198,13 @@ public CompletableFuture<Stat> storePut(String path, byte[] data, Optional<Long>
if (type == NotificationType.Created) {
notifyParentChildrenChanged(path);
}
return FutureUtils
.value(new Stat(path, newValue.version, newValue.createdTimestamp,
newValue.modifiedTimestamp,
false, true));
String finalPath = path;
execute(() -> future.complete(new Stat(finalPath, newValue.version, newValue.createdTimestamp,
newValue.modifiedTimestamp,
false, true)), future);
}
}
return future;
}
}

Expand All @@ -211,18 +214,20 @@ public CompletableFuture<Void> storeDelete(String path, Optional<Long> optExpect
return FutureUtil.failedFuture(new MetadataStoreException.InvalidPathException(path));
}
synchronized (map) {
CompletableFuture<Void> future = new CompletableFuture<>();
Value value = map.get(path);
if (value == null) {
return FutureUtils.exception(new NotFoundException(""));
execute(() -> future.completeExceptionally(new NotFoundException("")), future);
} else if (optExpectedVersion.isPresent() && optExpectedVersion.get() != value.version) {
return FutureUtils.exception(new BadVersionException(""));
execute(() -> future.completeExceptionally(new BadVersionException("")), future);
} else {
map.remove(path);
receivedNotification(new Notification(NotificationType.Deleted, path));

notifyParentChildrenChanged(path);
return FutureUtils.value(null);
execute(() -> future.complete(null), future);
}
return future;
}
}
}