public static Properties createPropsForRemoteStorage(String testClassName,
                                                         String storageDirPath,
                                                         int brokerCount,
                                                         int numRemoteLogMetadataPartitions,
                                                         Properties overridingProps) {
        Assertions.assertTrue(STORAGE_WAIT_TIMEOUT_SEC > TimeUnit.MILLISECONDS.toSeconds(RLM_TASK_INTERVAL_MS),
                "STORAGE_WAIT_TIMEOUT_SEC should be greater than RLM_TASK_INTERVAL_MS");

        // Configure the tiered storage in Kafka. Set an interval of 1 second for the remote log manager background
        // activity to ensure the tiered storage has enough room to be exercised within the lifetime of a test.
        //
        // The replication factor of the remote log metadata topic needs to be chosen so that in resiliency
        // tests, metadata can survive the loss of one replica for its topic-partitions.
        //
        // The second-tier storage system is mocked via the LocalTieredStorage instance which persists transferred
        // data files on the local file system.
        overridingProps.put(ServerConfigs.UNSTABLE_API_VERSIONS_ENABLE_CONFIG, "true");
        overridingProps.put(ServerConfigs.UNSTABLE_FEATURE_VERSIONS_ENABLE_CONFIG, "true");
        overridingProps.setProperty(REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, "true");
        overridingProps.setProperty(REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP, LocalTieredStorage.class.getName());
        overridingProps.setProperty(REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP,
                TopicBasedRemoteLogMetadataManager.class.getName());
        overridingProps.setProperty(REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_PROP, RLM_TASK_INTERVAL_MS.toString());
        overridingProps.setProperty(REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_PROP, "PLAINTEXT");

        overridingProps.setProperty(REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_PROP, storageConfigPrefix(testClassName, ""));
        overridingProps.setProperty(REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_PROP, metadataConfigPrefix(testClassName, ""));

        overridingProps.setProperty(
                metadataConfigPrefix(testClassName, TopicBasedRemoteLogMetadataManagerConfig.REMOTE_LOG_METADATA_TOPIC_PARTITIONS_PROP),
                String.valueOf(numRemoteLogMetadataPartitions));
        overridingProps.setProperty(
                metadataConfigPrefix(testClassName, TopicBasedRemoteLogMetadataManagerConfig.REMOTE_LOG_METADATA_TOPIC_REPLICATION_FACTOR_PROP),
                String.valueOf(brokerCount));
        // The below two configurations ensures inactive log segments are deleted fast enough so that
        // the integration tests can confirm a given log segment is present only in the second-tier storage.
        // Note that this does not impact the eligibility of a log segment to be offloaded to the
        // second-tier storage.
        overridingProps.setProperty(LOG_INITIAL_TASK_DELAY_MS_CONFIG, LOG_CLEANUP_INTERVAL_MS.toString());
        overridingProps.setProperty(LOG_CLEANUP_INTERVAL_MS_CONFIG, LOG_CLEANUP_INTERVAL_MS.toString());
        // The directory of the second-tier storage needs to be constant across all instances of storage managers
        // in every broker and throughout the test. Indeed, as brokers are restarted during the test.
        // You can override this property with a fixed path of your choice if you wish to use a non-temporary
        // directory to access its content after a test terminated.
        overridingProps.setProperty(storageConfigPrefix(testClassName, STORAGE_DIR_CONFIG), storageDirPath);
        // This configuration will remove all the remote files when close is called in remote storage manager.
        // Storage manager close is being called while the server is actively processing the socket requests,
        // so enabling this config can break the existing tests.
        // NOTE: When using TestUtils#tempDir(), the folder gets deleted when VM terminates.
        overridingProps.setProperty(storageConfigPrefix(testClassName, DELETE_ON_CLOSE_CONFIG), "false");
        // Set a small number of retry interval for retrying RemoteLogMetadataManager resources initialization to speed up the test
        overridingProps.setProperty(metadataConfigPrefix(testClassName, REMOTE_LOG_METADATA_INITIALIZATION_RETRY_INTERVAL_MS_PROP), RLMM_INIT_RETRY_INTERVAL_MS.toString());
        // Set 2 log dirs to make sure JBOD feature is working correctly
        overridingProps.setProperty(ServerLogConfigs.LOG_DIRS_CONFIG, TestUtils.tempDir().getAbsolutePath() + "," + TestUtils.tempDir().getAbsolutePath());
        // Disable unnecessary log cleaner
        overridingProps.setProperty(LOG_CLEANER_ENABLE_PROP, "false");

        return overridingProps;
    }
Get the records found in the local tiered storage.
Snapshot does not sort the filesets by base offset.
@param context The test context.
@param topicPartition The topic-partition of the records.
@return The records found in the local tiered storage.
storage/src/test/java/org/apache/kafka/tiered/storage/utils/TieredStorageTestUtils.java