Repository: apache/kafka
File: storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManager.java (Java)

Code before:
                } finally {
                    lock.writeLock().unlock();
                }
            }
        } catch (Exception e) {
            log.error("Encountered error while initializing topic-based RLMM resources", e);
            initializationFailed = true;
        } finally {
            Utils.closeQuietly(adminClient, "AdminClient");
        }
    }

    boolean doesTopicExist(Admin adminClient, String topic) throws ExecutionException, InterruptedException {
        try {
            TopicDescription description = adminClient.describeTopics(Set.of(topic))
                    .topicNameValues()
                    .get(topic)
                    .get();
            if (description != null) {
                log.info("Topic {} exists. TopicId: {}, numPartitions: {}, ", topic,
                        description.topicId(), description.partitions().size());
            }
            return true;
        } catch (ExecutionException | InterruptedException ex) {
            if (ex.getCause() instanceof UnknownTopicOrPartitionException) {
                log.info("Topic {} does not exist", topic);
                return false;
            }
            log.warn("Fail to check if topic {} exist. Error: {}", topic, ex.getCause().getMessage());
            throw ex;
        }
    }

    private boolean isPartitionsCountSameAsConfigured(Admin adminClient,
                                                      String topicName) throws InterruptedException, ExecutionException {
        log.debug("Getting topic details to check for partition count and replication factor.");
        TopicDescription topicDescription = adminClient.describeTopics(Set.of(topicName))
                                                       .topicNameValues().get(topicName).get();
        int expectedPartitions = rlmmConfig.metadataTopicPartitionsCount();
        int topicPartitionsSize = topicDescription.partitions().size();

        if (topicPartitionsSize != expectedPartitions) {
            log.error("Existing topic partition count [{}] is not same as the expected partition count [{}]",
                      topicPartitionsSize, expectedPartitions);
            return false;
        }

        return true;
    }

    private NewTopic createRemoteLogMetadataTopicRequest() {

Review comment (bug):
let me give some detailed information: 
when topic doesn't existed. the exception will be throw with cause: UnknownTopicOrPartitionException
I test it both on 3.9.0 and 4.0.0.
The normal "exception" throwed on restarting process don't give the conclusion that the topic isn't existed. 
most of the cases are the connection fail to connect due to the server isn't ready.

Code after:
                } finally {
                    lock.writeLock().unlock();
                }
            }
        } catch (Exception e) {
            log.error("Encountered error while initializing topic-based RLMM resources", e);
            initializationFailed = true;
        } finally {
            Utils.closeQuietly(adminClient, "AdminClient");
        }
    }

    boolean doesTopicExist(Admin adminClient, String topic) throws ExecutionException, InterruptedException {
        try {
            TopicDescription description = adminClient.describeTopics(Set.of(topic))
                    .topicNameValues()
                    .get(topic)
                    .get();
            if (description != null) {
                log.info("Topic {} exists. TopicId: {}, numPartitions: {}, ", topic,
                        description.topicId(), description.partitions().size());
            }
            return true;
        } catch (ExecutionException | InterruptedException ex) {
            if (ex.getCause() instanceof UnknownTopicOrPartitionException) {
                log.info("Topic {} does not exist", topic);
                return false;
            }
            throw ex;
        }
    }

    private boolean isPartitionsCountSameAsConfigured(Admin adminClient,
                                                      String topicName) throws InterruptedException, ExecutionException {
        log.debug("Getting topic details to check for partition count and replication factor.");
        TopicDescription topicDescription = adminClient.describeTopics(Set.of(topicName))
                                                       .topicNameValues().get(topicName).get();
        int expectedPartitions = rlmmConfig.metadataTopicPartitionsCount();
        int topicPartitionsSize = topicDescription.partitions().size();

        if (topicPartitionsSize != expectedPartitions) {
            log.error("Existing topic partition count [{}] is not same as the expected partition count [{}]",
                      topicPartitionsSize, expectedPartitions);
            return false;
        }

        return true;
    }

    private NewTopic createRemoteLogMetadataTopicRequest() {
        Map<String, String> topicConfigs = new HashMap<>();