Skip to content

Commit

Permalink
[fix] [meta] Oxia metadta store: Convert error to MetadataStoreExcept…
Browse files Browse the repository at this point in the history
…ion if operation failed (apache#23154)
  • Loading branch information
poorbarcode authored Aug 12, 2024
1 parent 1b43b9d commit 38134bc
Show file tree
Hide file tree
Showing 2 changed files with 60 additions and 6 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.metadata.api.GetResult;
import org.apache.pulsar.metadata.api.MetadataEventSynchronizer;
import org.apache.pulsar.metadata.api.MetadataStoreConfig;
Expand Down Expand Up @@ -252,14 +253,16 @@ protected CompletableFuture<Stat> storePut(
}

private <T> CompletionStage<T> convertException(Throwable ex) {
if (ex.getCause() instanceof UnexpectedVersionIdException
|| ex.getCause() instanceof KeyAlreadyExistsException) {
Throwable actEx = FutureUtil.unwrapCompletionException(ex);
if (actEx instanceof UnexpectedVersionIdException || actEx instanceof KeyAlreadyExistsException) {
return CompletableFuture.failedFuture(
new MetadataStoreException.BadVersionException(ex.getCause()));
} else if (ex.getCause() instanceof IllegalStateException) {
return CompletableFuture.failedFuture(new MetadataStoreException.AlreadyClosedException(ex.getCause()));
new MetadataStoreException.BadVersionException(actEx));
} else if (actEx instanceof IllegalStateException) {
return CompletableFuture.failedFuture(new MetadataStoreException.AlreadyClosedException(actEx));
} else if (actEx instanceof MetadataStoreException) {
return CompletableFuture.failedFuture(actEx);
} else {
return CompletableFuture.failedFuture(ex.getCause());
return CompletableFuture.failedFuture(new MetadataStoreException(actEx));
}
}

Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you 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 org.apache.pulsar.metadata;

import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.metadata.api.MetadataStore;
import org.apache.pulsar.metadata.api.MetadataStoreConfig;
import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.metadata.api.MetadataStoreFactory;
import org.testng.annotations.Test;

@Slf4j
public class OxiaMetadataStoreErrorTest extends BaseMetadataStoreTest {

@Test
public void emptyStoreTest() throws Exception {
String metadataStoreUrl = "oxia://" + getOxiaServerConnectString();
String prefix = newKey();
@Cleanup
MetadataStore store = MetadataStoreFactory.create(metadataStoreUrl,
MetadataStoreConfig.builder().fsyncEnable(false).build());
oxiaServer.close();
try {
store.exists(prefix + "/non-existing-key").join();
fail("Expected an exception because the metadata store server has been closed.");
} catch (Exception ex) {
Throwable actEx = FutureUtil.unwrapCompletionException(ex);
assertTrue(actEx instanceof MetadataStoreException);
}
}
}

0 comments on commit 38134bc

Please sign in to comment.