You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2024/06/06 07:08:49 UTC
(pulsar) branch branch-3.1 updated: [fix][meta] Check if metadata store is closed in RocksdbMetadataStore (#22852)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.1 by this push:
new 396e3e62be7 [fix][meta] Check if metadata store is closed in RocksdbMetadataStore (#22852)
396e3e62be7 is described below
commit 396e3e62be76c71f1b478d47887d3b4b5143f3e6
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Wed Jun 5 22:02:43 2024 +0300
[fix][meta] Check if metadata store is closed in RocksdbMetadataStore (#22852)
(cherry picked from commit 74192871ed00870e5181a5bd4018ba196fd8f698)
---
.../metadata/impl/AbstractMetadataStore.java | 25 +++++++++++-----------
.../pulsar/metadata/impl/RocksdbMetadataStore.java | 15 +++++++++++++
2 files changed, 27 insertions(+), 13 deletions(-)
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java
index 4cadf2397a7..18529f82cac 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java
@@ -246,8 +246,7 @@ public abstract class AbstractMetadataStore implements MetadataStoreExtended, Co
@Override
public CompletableFuture<Optional<GetResult>> get(String path) {
if (isClosed()) {
- return FutureUtil.failedFuture(
- new MetadataStoreException.AlreadyClosedException());
+ return alreadyClosedFailedFuture();
}
long start = System.currentTimeMillis();
if (!isValidPath(path)) {
@@ -275,8 +274,7 @@ public abstract class AbstractMetadataStore implements MetadataStoreExtended, Co
@Override
public final CompletableFuture<List<String>> getChildren(String path) {
if (isClosed()) {
- return FutureUtil.failedFuture(
- new MetadataStoreException.AlreadyClosedException());
+ return alreadyClosedFailedFuture();
}
if (!isValidPath(path)) {
return FutureUtil.failedFuture(new MetadataStoreException.InvalidPathException(path));
@@ -287,8 +285,7 @@ public abstract class AbstractMetadataStore implements MetadataStoreExtended, Co
@Override
public final CompletableFuture<Boolean> exists(String path) {
if (isClosed()) {
- return FutureUtil.failedFuture(
- new MetadataStoreException.AlreadyClosedException());
+ return alreadyClosedFailedFuture();
}
if (!isValidPath(path)) {
return FutureUtil.failedFuture(new MetadataStoreException.InvalidPathException(path));
@@ -350,8 +347,7 @@ public abstract class AbstractMetadataStore implements MetadataStoreExtended, Co
@Override
public final CompletableFuture<Void> delete(String path, Optional<Long> expectedVersion) {
if (isClosed()) {
- return FutureUtil.failedFuture(
- new MetadataStoreException.AlreadyClosedException());
+ return alreadyClosedFailedFuture();
}
long start = System.currentTimeMillis();
if (!isValidPath(path)) {
@@ -400,8 +396,7 @@ public abstract class AbstractMetadataStore implements MetadataStoreExtended, Co
@Override
public CompletableFuture<Void> deleteRecursive(String path) {
if (isClosed()) {
- return FutureUtil.failedFuture(
- new MetadataStoreException.AlreadyClosedException());
+ return alreadyClosedFailedFuture();
}
return getChildren(path)
.thenCompose(children -> FutureUtil.waitForAll(
@@ -425,8 +420,7 @@ public abstract class AbstractMetadataStore implements MetadataStoreExtended, Co
public final CompletableFuture<Stat> put(String path, byte[] data, Optional<Long> optExpectedVersion,
EnumSet<CreateOption> options) {
if (isClosed()) {
- return FutureUtil.failedFuture(
- new MetadataStoreException.AlreadyClosedException());
+ return alreadyClosedFailedFuture();
}
long start = System.currentTimeMillis();
if (!isValidPath(path)) {
@@ -506,10 +500,15 @@ public abstract class AbstractMetadataStore implements MetadataStoreExtended, Co
}
}
- private boolean isClosed() {
+ protected boolean isClosed() {
return isClosed.get();
}
+ protected static <T> CompletableFuture<T> alreadyClosedFailedFuture() {
+ return FutureUtil.failedFuture(
+ new MetadataStoreException.AlreadyClosedException());
+ }
+
@Override
public void close() throws Exception {
executor.shutdownNow();
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/RocksdbMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/RocksdbMetadataStore.java
index fcec2fcf9c1..0187a3fdac1 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/RocksdbMetadataStore.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/RocksdbMetadataStore.java
@@ -376,6 +376,9 @@ public class RocksdbMetadataStore extends AbstractMetadataStore {
}
try {
dbStateLock.readLock().lock();
+ if (isClosed()) {
+ return alreadyClosedFailedFuture();
+ }
byte[] value = db.get(optionCache, toBytes(path));
if (value == null) {
return CompletableFuture.completedFuture(Optional.empty());
@@ -408,6 +411,9 @@ public class RocksdbMetadataStore extends AbstractMetadataStore {
}
try {
dbStateLock.readLock().lock();
+ if (isClosed()) {
+ return alreadyClosedFailedFuture();
+ }
try (RocksIterator iterator = db.newIterator(optionDontCache)) {
Set<String> result = new HashSet<>();
String firstKey = path.equals("/") ? path : path + "/";
@@ -450,6 +456,9 @@ public class RocksdbMetadataStore extends AbstractMetadataStore {
}
try {
dbStateLock.readLock().lock();
+ if (isClosed()) {
+ return alreadyClosedFailedFuture();
+ }
byte[] value = db.get(optionDontCache, toBytes(path));
if (log.isDebugEnabled()) {
if (value != null) {
@@ -472,6 +481,9 @@ public class RocksdbMetadataStore extends AbstractMetadataStore {
}
try {
dbStateLock.readLock().lock();
+ if (isClosed()) {
+ return alreadyClosedFailedFuture();
+ }
try (Transaction transaction = db.beginTransaction(writeOptions)) {
byte[] pathBytes = toBytes(path);
byte[] oldValueData = transaction.getForUpdate(optionDontCache, pathBytes, true);
@@ -508,6 +520,9 @@ public class RocksdbMetadataStore extends AbstractMetadataStore {
}
try {
dbStateLock.readLock().lock();
+ if (isClosed()) {
+ return alreadyClosedFailedFuture();
+ }
try (Transaction transaction = db.beginTransaction(writeOptions)) {
byte[] pathBytes = toBytes(path);
byte[] oldValueData = transaction.getForUpdate(optionDontCache, pathBytes, true);