@Override
    public void configure(Map<String, ?> configs) {
        if (storageDirectory != null) {
            throw new InvalidConfigurationException(format("This instance of local remote storage" +
                    "is already configured. The existing storage directory is %s. Ensure the method " +
                    "configure() is only called once.", storageDirectory.getAbsolutePath()));
        }

        final String storageDir = (String) configs.get(STORAGE_DIR_CONFIG);
        final String shouldDeleteOnClose = (String) configs.get(DELETE_ON_CLOSE_CONFIG);
        final String transfererClass = (String) configs.get(TRANSFERER_CLASS_CONFIG);
        final String isDeleteEnabled = (String) configs.get(ENABLE_DELETE_API_CONFIG);
        final Integer brokerIdInt = (Integer) configs.get(BROKER_ID);

        if (brokerIdInt == null) {
            throw new InvalidConfigurationException(
                    "Broker ID is required to configure the LocalTieredStorage manager.");
        }

        brokerId = brokerIdInt;
        logger = new LogContext(format("[LocalTieredStorage Id=%d] ", brokerId)).logger(this.getClass());

        if (shouldDeleteOnClose != null) {
            deleteOnClose = Boolean.parseBoolean(shouldDeleteOnClose);
        }

        if (isDeleteEnabled != null) {
            deleteEnabled = Boolean.parseBoolean(isDeleteEnabled);
        }

        if (transfererClass != null) {
            try {
                transferer = (Transferer) getClass().getClassLoader().loadClass(transfererClass).getDeclaredConstructor().newInstance();

            } catch (InstantiationException | IllegalAccessException | ClassNotFoundException | ClassCastException | InvocationTargetException | NoSuchMethodException e) {
                throw new RuntimeException(format("Cannot create transferer from class '%s'", transfererClass), e);
            }
        }


        if (storageDir == null) {
            storageDirectory = TestUtils.tempDirectory(ROOT_STORAGE_DIR_NAME + "-");

            logger.debug("No storage directory specified, created temporary directory: {}",
                    storageDirectory.getAbsolutePath());

        } else {
            // LocalTieredStorage implementation can be used in two modes:
            // 1. Integration test and
            // 2. To bring Kafka in the local machine with LocalTieredStorage implementation as RSM for quick testing
            //    by adding kafka-storage-<version>-test.jar in the classpath, then enabling the INCLUDE_TEST_JARS flag
            //    before running the server
            //
            // In mode-2, if the commented-out #L284 is enabled, then it creates the storageDirectory from the present
            // kafka working directory which can cause confusion:
            //
            // (eg) Kafka is installed on /opt/kafka and storageDirectory is configured as /tmp/kafka-tiered-storage,
            // then it will create it the directory in /opt/kafka/kafka-tiered-storage/tmp/kafka-tiered-storage instead
            //  of /tmp/kafka-tiered-storage.
            //
            // storageDirectory = new File(new File("."), ROOT_STORAGES_DIR_NAME + "/" + storageDir);
            storageDirectory = new File(storageDir, ROOT_STORAGE_DIR_NAME);
            final boolean existed = Files.exists(storageDirectory.toPath());
            if (!existed) {
                try {
                    logger.info("Creating directory: [{}]", storageDirectory.getAbsolutePath());
                    Files.createDirectories(storageDirectory.toPath());
                } catch (IOException e) {
                    throw new RuntimeException(format("Not able to create the storage directory '%s'", storageDirectory.getAbsolutePath()), e);
                }
            } else {
                logger.warn("Remote storage with ID [{}] already exists on the file system. Any data already " +
                        "in the remote storage will not be deleted and may result in an inconsistent state and/or " +
                        "provide stale data.", storageDir);
            }
        }

        logger.info("Created local tiered storage manager [{}]:[{}]", brokerId, storageDirectory.getName());
    }
Walks through this storage and notify the traverser of every topic-partition, segment and record discovered.
<p>
- The order of traversal of the topic-partition is not specified.
- The order of traversal of the segments within a topic-partition is in ascending order
of the modified timestamp of the segment file.
- The order of traversal of records within a segment corresponds to the insertion
order of these records in the original segment from which the segment in this storage
was transferred from.
<p>
This method is NOT an atomic operation w.r.t the local tiered storage. This storage may change while
being traversed topic-partitions, segments and records are communicated to the traverser. There is
no guarantee updates to the storage which happens during traversal will be communicated to the traverser.
Especially, in case of concurrent read and write/delete to a topic-partition, a segment or a record,
the behaviour depends on the underlying file system.

@param traverser User-specified object to which this storage communicates the topic-partitions,
                 segments and records as they are discovered.
storage/src/test/java/org/apache/kafka/server/log/remote/storage/LocalTieredStorage.java