kamalcph commented on code in PR #14116: URL: https://github.com/apache/kafka/pull/14116#discussion_r1294640475
########## storage/src/test/java/org/apache/kafka/tiered/storage/TieredStorageTestHarness.java: ########## @@ -0,0 +1,204 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.tiered.storage; + +import org.apache.kafka.tiered.storage.utils.BrokerLocalStorage; +import kafka.api.IntegrationTestHarness; +import kafka.log.remote.RemoteLogManager; +import kafka.server.KafkaBroker; +import kafka.server.KafkaConfig; +import kafka.utils.TestUtils; +import org.apache.kafka.common.replica.ReplicaSelector; +import org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManager; +import org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManagerConfig; +import org.apache.kafka.server.log.remote.storage.ClassLoaderAwareRemoteStorageManager; +import org.apache.kafka.server.log.remote.storage.LocalTieredStorage; +import org.apache.kafka.server.log.remote.storage.RemoteStorageManager; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInfo; +import scala.collection.Seq; + +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.Properties; +import java.util.stream.Collectors; + +import scala.collection.JavaConverters; + +import static org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP; +import static org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig.REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP; +import static org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig.REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP; +import static org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig.REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_PROP; + +import static org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig.REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_PROP; +import static org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig.REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_PROP; + +import static org.apache.kafka.server.log.remote.storage.LocalTieredStorage.DELETE_ON_CLOSE_CONFIG; +import static org.apache.kafka.server.log.remote.storage.LocalTieredStorage.STORAGE_DIR_CONFIG; + +/** + * Base class for integration tests exercising the tiered storage functionality in Apache Kafka. + */ +@Tag("integration") +public abstract class TieredStorageTestHarness extends IntegrationTestHarness { + + /** + * InitialTaskDelayMs is set to 30 seconds for the delete-segment scheduler in Apache Kafka. + * Hence, we need to wait at least that amount of time before segments eligible for deletion + * gets physically removed. + */ + private static final Integer STORAGE_WAIT_TIMEOUT_SEC = 35; + + private TieredStorageTestContext context; + + protected int numRemoteLogMetadataPartitions() { + return 5; + } + + @SuppressWarnings("deprecation") + @Override + public void modifyConfigs(Seq<Properties> props) { + for (Properties p : JavaConverters.seqAsJavaList(props)) { + p.putAll(overridingProps()); + } + } + + public Properties overridingProps() { + Properties overridingProps = new Properties(); + // 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.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, "1000"); + + overridingProps.setProperty(REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_PROP, storageConfigPrefix("")); Review Comment: Used the test class name in the prefix instead of display name as it always be "executeTieredStorageTest". -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
