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

Code before:
                    continue;
                }
                // Create producer and consumer managers.
                lock.writeLock().lock();
                try {
                    producerManager = new ProducerManager(rlmmConfig, partitioner);
                    consumerManager = new ConsumerManager(rlmmConfig, remotePartitionMetadataStore, partitioner, time);
                    consumerManager.startConsumerThread();
                    if (!pendingAssignPartitions.isEmpty()) {
                        assignPartitions(pendingAssignPartitions);
                        pendingAssignPartitions.clear();
                    }
                    initialized.set(true);
                    log.info("Initialized topic-based RLMM resources successfully");
                } catch (Exception e) {
                    log.error("Encountered error while initializing producer/consumer", e);
                    initializationFailed = true;
                } finally {
                    lock.writeLock().unlock();
                }
            }
        } catch (KafkaException e) {
            log.error("Encountered error while initializing topic-based RLMM resources", e);
            initializationFailed = true;
        } finally {
            Utils.closeQuietly(adminClient, "AdminClient");
            if (initializationFailed) {
                log.error("Stopping the server as it failed to initialize topic-based RLMM resources");
                Exit.exit(1);
            }
        }
    }

    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;
        }
    }


Review comment (bug):
The `initializationFailed` variable can be moved from global to local scope.

Code after:
                    continue;
                }
                // Create producer and consumer managers.
                lock.writeLock().lock();
                try {
                    producerManager = new ProducerManager(rlmmConfig, partitioner);
                    consumerManager = new ConsumerManager(rlmmConfig, remotePartitionMetadataStore, partitioner, time);
                    consumerManager.startConsumerThread();
                    if (!pendingAssignPartitions.isEmpty()) {
                        assignPartitions(pendingAssignPartitions);
                        pendingAssignPartitions.clear();
                    }
                    initialized.set(true);
                    log.info("Initialized topic-based RLMM resources successfully");
                } catch (Exception e) {
                    log.error("Encountered error while initializing producer/consumer", e);
                    initializationFailed = true;
                } finally {
                    lock.writeLock().unlock();
                }
            }
        } catch (KafkaException e) {
            log.error("Encountered error while initializing topic-based RLMM resources", e);
            initializationFailed = true;
        } finally {
            if (initializationFailed) {
                log.error("Stopping the server as it failed to initialize topic-based RLMM resources");
                Exit.exit(1);
            }
        }
    }

    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,