Fix topic lookup failed exceptionally with ServiceUnitNotReady

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaTopicLookupService.java

@@ -15,6 +15,7 @@
 
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
 import lombok.NonNull;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.broker.service.BrokerService;
@@ -96,8 +97,9 @@ private void handleGetTopicException(@NonNull final String topicName,
                                          final CompletableFuture<Optional<PersistentTopic>> topicCompletableFuture,
                                          @NonNull final Throwable ex,
                                          @NonNull final Object requestor) {
+        final Throwable realThrowable = (ex instanceof CompletionException) ? ex.getCause() : ex;
         // The ServiceUnitNotReadyException is retryable, so we should print a warning log instead of error log
-        if (ex instanceof BrokerServiceException.ServiceUnitNotReadyException) {
+        if (realThrowable instanceof BrokerServiceException.ServiceUnitNotReadyException) {
             log.warn("[{}] Failed to getTopic {}: {}",
                     requestor, topicName, ex.getMessage());
             topicCompletableFuture.complete(Optional.empty());