KAFKA-17689: Remove unused TieredStorageTestHarness

storage/src/test/java/org/apache/kafka/tiered/storage/README.md

@@ -1,11 +1,11 @@
 # The Test Flow
 
-Step 1: For every test, setup is done via TieredStorageTestHarness which extends IntegrationTestHarness and sets up a cluster with TS enabled on it.
+Step 1: Each test is a standalone class. It declares a `clusterConfig()` method that returns a `List<ClusterConfig>` with tiered storage enabled (via `TieredStorageTestUtils.createServerPropsForRemoteStorage`), and test methods annotated with `@ClusterTemplate("clusterConfig")` that receive a `ClusterInstance` provided by the test framework.
 
-Step 2: The test is written as a specification consisting of sequential actions and assertions. The spec for the complete test is written down first which creates "actions" to be executed.
+Step 2: The test is written as a specification consisting of sequential actions and assertions. The spec for the complete test is built first using `TieredStorageTestBuilder`, which creates the "actions" to be executed.
 
-Step 3: Once we have the test spec in-place (which includes assertion actions), we execute the test which will execute each action sequentially. 
+Step 3: A `TieredStorageTestContext` is created from the `ClusterInstance` (plus any extra consumer config). The test is then executed by running each action of the spec sequentially against the context.
 
-Step 4: The test execution stops when any of the action throws an exception (or an assertion error).
+Step 4: The test execution stops when any of the actions throws an exception (or an assertion error).
 
-Step 5: Clean-up for the test is performed on test exit
+Step 5: Clean-up for the test is performed on test exit — the `TieredStorageTestContext` is closed (it is `AutoCloseable`, typically via try-with-resources) and the test report is printed.

storage/src/test/java/org/apache/kafka/tiered/storage/TieredStorageTestContext.java

@@ -16,6 +16,8 @@
  */
 package org.apache.kafka.tiered.storage;
 
+import kafka.server.KafkaBroker;
+
 import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.clients.admin.AlterConfigOp;
 import org.apache.kafka.clients.admin.AlterConfigsOptions;
@@ -37,9 +39,12 @@
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.apache.kafka.common.test.ClusterInstance;
 import org.apache.kafka.common.utils.Utils;
+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.LocalTieredStorageHistory;
 import org.apache.kafka.server.log.remote.storage.LocalTieredStorageSnapshot;
+import org.apache.kafka.server.log.remote.storage.RemoteLogManager;
+import org.apache.kafka.server.log.remote.storage.RemoteStorageManager;
 import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache;
 import org.apache.kafka.storage.internals.log.UnifiedLog;
 import org.apache.kafka.test.TestUtils;
@@ -57,12 +62,14 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.stream.Collectors;
 
 import static org.apache.kafka.clients.producer.ProducerConfig.LINGER_MS_CONFIG;
+import static org.apache.kafka.tiered.storage.utils.TieredStorageTestUtils.STORAGE_WAIT_TIMEOUT_SEC;
 
 public final class TieredStorageTestContext implements AutoCloseable {
 
@@ -104,8 +111,36 @@ private void initClients() {
     }
 
     private void initContext() {
-        remoteStorageManagers = TieredStorageTestHarness.remoteStorageManagers(cluster.aliveBrokers().values());
-        localStorages = TieredStorageTestHarness.localStorages(cluster.aliveBrokers().values());
+        remoteStorageManagers = remoteStorageManagers(cluster.aliveBrokers().values());
+        localStorages = localStorages(cluster.aliveBrokers().values());
+    }
+
+    private static List<LocalTieredStorage> remoteStorageManagers(Collection<KafkaBroker> brokers) {
+        List<LocalTieredStorage> storages = new ArrayList<>();
+        brokers.forEach(broker -> {
+            if (broker.remoteLogManagerOpt().isDefined()) {
+                RemoteLogManager remoteLogManager = broker.remoteLogManagerOpt().get();
+                RemoteStorageManager storageManager = remoteLogManager.storageManager();
+                if (storageManager instanceof ClassLoaderAwareRemoteStorageManager loaderAwareRSM) {
+                    if (loaderAwareRSM.delegate() instanceof LocalTieredStorage) {
+                        storages.add((LocalTieredStorage) loaderAwareRSM.delegate());
+                    }
+                } else if (storageManager instanceof LocalTieredStorage) {
+                    storages.add((LocalTieredStorage) storageManager);
+                }
+            } else {
+                throw new AssertionError("Broker " + broker.config().brokerId()
+                        + " does not have a remote log manager.");
+            }
+        });
+        return storages;
+    }
+
+    private static List<BrokerLocalStorage> localStorages(Collection<KafkaBroker> brokers) {
+        return brokers.stream()
+                .map(b -> new BrokerLocalStorage(b.config().brokerId(), Set.copyOf(b.config().logDirs()),
+                        STORAGE_WAIT_TIMEOUT_SEC))
+                .toList();
     }
 
     public void createTopic(TopicSpec spec) throws ExecutionException, InterruptedException {
@@ -260,7 +295,7 @@ public void eraseBrokerStorage(int brokerId,
                                    boolean isStopped) throws IOException {
         BrokerLocalStorage brokerLocalStorage;
         if (isStopped) {
-            brokerLocalStorage = TieredStorageTestHarness.localStorages(cluster.brokers().values())
+            brokerLocalStorage = localStorages(cluster.brokers().values())
                     .stream()
                     .filter(bls -> bls.getBrokerId() == brokerId)
                     .findFirst()