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

Code before:
            initializationFailed = true;
        } finally {
            if (initializationFailed) {
                log.error("Stopping the server as it failed to initialize topic-based RLMM resources");
                Exit.exit(1);
            }
        }
    }

    /**
     * Invoked when the broker is ready to handle requests. This triggers the initialization of
     * resources including Kafka clients that operate on the remote log metadata topic.
     * <p>
     * The target cluster for the topic is determined by configuration and can be either:
     * <ol>
     *   <li>The local cluster (most common) - the initialization is deferred until the broker is ready
     *       to handle requests. Early initialization would lead to connection failures.</li>
     *   <li>A remote cluster - the delay is not necessary but causes no harm.</li>
     * </ol>
     * <p>
     * By using the broker ready state as the initialization trigger, the implementation optimally handles
     * the typical case while remaining correct for alternative configurations.
     */
    @Override
    public void onBrokerReady() {
        log.info("Broker is ready for requests, now initializing topic-based RLMM resources");
        initializationThread.start();
    }

    boolean doesTopicExist(Admin admin, String topic) throws ExecutionException, InterruptedException {
        try {
            TopicDescription description = admin.describeTopics(Set.of(topic))
                    .topicNameValues()
                    .get(topic)
                    .get();
            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 admin,
                                                      String topicName,
                                                      int metadataTopicPartitionCount) throws InterruptedException, ExecutionException {
        log.debug("Getting topic details to check for partition count and replication factor.");

Review comment (refactor):
Since `initializeResources` already logs `Initializing topic-based RLMM resources`. This duplicate log may not be necessary

Code after:
            initializationFailed = true;
        } finally {
            if (initializationFailed) {
                log.error("Stopping the server as it failed to initialize topic-based RLMM resources");
                Exit.exit(1);
            }
        }
    }

    /**
     * Invoked when the broker is ready to handle requests. This triggers the initialization of
     * resources including Kafka clients that operate on the remote log metadata topic.
     * <p>
     * The target cluster for the topic is determined by configuration and can be either:
     * <ol>
     *   <li>The local cluster (most common) - the initialization is deferred until the broker is ready
     *       to handle requests. Early initialization would lead to connection failures.</li>
     *   <li>A remote cluster - the delay is not necessary but causes no harm.</li>
     * </ol>
     * <p>
     * By using the broker ready state as the initialization trigger, the implementation optimally handles
     * the typical case while remaining correct for alternative configurations.
     */
    @Override
    public void onBrokerReady() {
        initializationThread.start();
    }

    boolean doesTopicExist(Admin admin, String topic) throws ExecutionException, InterruptedException {
        try {
            TopicDescription description = admin.describeTopics(Set.of(topic))
                    .topicNameValues()
                    .get(topic)
                    .get();
            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 admin,
                                                      String topicName,
                                                      int metadataTopicPartitionCount) throws InterruptedException, ExecutionException {
        log.debug("Getting topic details to check for partition count and replication factor.");
        TopicDescription topicDescription = admin