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

Code before:
     * Closes and releases all the resources like RemoterStorageManager and RemoteLogMetadataManager.
     */
    public void close() {
        synchronized (this) {
            if (!closed) {
                leaderCopyRLMTasks.values().forEach(RLMTaskWithFuture::cancel);
                leaderExpirationRLMTasks.values().forEach(RLMTaskWithFuture::cancel);
                followerRLMTasks.values().forEach(RLMTaskWithFuture::cancel);

                rlmCopyThreadPool.close();
                rlmExpirationThreadPool.close();
                followerThreadPool.close();
                try {
                    shutdownAndAwaitTermination(remoteStorageReaderThreadPool, "RemoteStorageReaderThreadPool", 10, TimeUnit.SECONDS);
                } finally {
                    removeMetrics();
                }
                leaderCopyRLMTasks.clear();
                leaderExpirationRLMTasks.clear();
                followerRLMTasks.clear();

                Utils.closeQuietly(indexCache, "RemoteIndexCache");
                Utils.closeQuietly(remoteLogMetadataManagerPlugin, "remoteLogMetadataManagerPlugin");
                Utils.closeQuietly(remoteStorageManagerPlugin, "remoteStorageManagerPlugin");
                Utils.closeQuietly(fetchQuotaMetrics, "fetchQuotaMetrics");
                Utils.closeQuietly(copyQuotaMetrics, "copyQuotaMetrics");
                closed = true;
            }
        }
    }

    private static void shutdownAndAwaitTermination(ExecutorService executor, String poolName, long timeout, TimeUnit timeUnit) {
        LOGGER.info("Shutting down {} executor", poolName);
        ThreadUtils.shutdownExecutorServiceQuietly(executor, timeout, timeUnit);
        LOGGER.info("{} executor shutdown completed", poolName);
    }

    //Visible for testing
    static ByteBuffer epochEntriesAsByteBuffer(List<EpochEntry> epochEntries) throws IOException {
        ByteArrayOutputStream stream = new ByteArrayOutputStream();
        try (BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(stream, StandardCharsets.UTF_8))) {
            CheckpointFile.CheckpointWriteBuffer<EpochEntry> writeBuffer =
                    new CheckpointFile.CheckpointWriteBuffer<>(writer, 0, LeaderEpochCheckpointFile.FORMATTER);
            writeBuffer.write(epochEntries);
            writer.flush();
        }

        return ByteBuffer.wrap(stream.toByteArray());
    }

    private void removeRemoteTopicPartitionMetrics(TopicIdPartition topicIdPartition) {

Review comment (suggestion):
minor suggestions: 

shall we move these 2 lines to RemoteLogManager#removeMetrics() method? And, call RemoteLogManager#removeMetrics()  at the end after closing `remoteStorageManagerPlugin`. 

Also, can the try..finally block be updated? 

```java
public void close() {
    synchronized (this) {
        if (!closed) {
          try {
             ..
             ..
             closed = true;
          } finally {
            removeMetrics();
         }
    }
}
```

Code after:

    /**
     * Closes and releases all the resources like RemoterStorageManager and RemoteLogMetadataManager.
     */
    public void close() {
        synchronized (this) {
            if (!closed) {
                leaderCopyRLMTasks.values().forEach(RLMTaskWithFuture::cancel);
                leaderExpirationRLMTasks.values().forEach(RLMTaskWithFuture::cancel);
                followerRLMTasks.values().forEach(RLMTaskWithFuture::cancel);

                rlmCopyThreadPool.close();
                rlmExpirationThreadPool.close();
                followerThreadPool.close();
                try {
                    shutdownAndAwaitTermination(remoteStorageReaderThreadPool, "RemoteStorageReaderThreadPool", 10, TimeUnit.SECONDS);

                    leaderCopyRLMTasks.clear();
                    leaderExpirationRLMTasks.clear();
                    followerRLMTasks.clear();

                    Utils.closeQuietly(indexCache, "RemoteIndexCache");
                    Utils.closeQuietly(remoteLogMetadataManagerPlugin, "remoteLogMetadataManagerPlugin");
                    Utils.closeQuietly(remoteStorageManagerPlugin, "remoteStorageManagerPlugin");
                    closed = true;
                } finally {
                    removeMetrics();
                }
            }
        }
    }

    private static void shutdownAndAwaitTermination(ExecutorService executor, String poolName, long timeout, TimeUnit timeUnit) {
        LOGGER.info("Shutting down {} executor", poolName);
        ThreadUtils.shutdownExecutorServiceQuietly(executor, timeout, timeUnit);
        LOGGER.info("{} executor shutdown completed", poolName);
    }

    //Visible for testing
    static ByteBuffer epochEntriesAsByteBuffer(List<EpochEntry> epochEntries) throws IOException {
        ByteArrayOutputStream stream = new ByteArrayOutputStream();
        try (BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(stream, StandardCharsets.UTF_8))) {
            CheckpointFile.CheckpointWriteBuffer<EpochEntry> writeBuffer =
                    new CheckpointFile.CheckpointWriteBuffer<>(writer, 0, LeaderEpochCheckpointFile.FORMATTER);
            writeBuffer.write(epochEntries);
            writer.flush();
        }

        return ByteBuffer.wrap(stream.toByteArray());
    }