fix handling of shared container

Author: dlg99

tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java

@@ -72,13 +72,7 @@ public class PulsarCluster {
      * @return the built pulsar cluster
      */
     public static PulsarCluster forSpec(PulsarClusterSpec spec) {
-        CSContainer csContainer = null;
-        if (!spec.enableOxia) {
-            csContainer = new CSContainer(spec.clusterName)
-                    .withNetwork(Network.newNetwork())
-                    .withNetworkAliases(CSContainer.NAME);
-        }
-        return new PulsarCluster(spec, null, csContainer, false);
+        return forSpec(spec, Network.newNetwork());
     }
 
     public static PulsarCluster forSpec(PulsarClusterSpec spec, Network network) {
@@ -93,7 +87,7 @@ public static PulsarCluster forSpec(PulsarClusterSpec spec, Network network) {
     }
 
     public static PulsarCluster forSpec(PulsarClusterSpec spec, CSContainer csContainer) {
-        return new PulsarCluster(spec, null, csContainer, true);
+        return new PulsarCluster(spec, csContainer.getNetwork(), csContainer, true);
     }
 
     @Getter
@@ -121,13 +115,12 @@ public static PulsarCluster forSpec(PulsarClusterSpec spec, CSContainer csContai
     private final String configurationMetadataStoreUrl;
 
     private PulsarCluster(PulsarClusterSpec spec, Network network, CSContainer csContainer, boolean sharedCsContainer) {
-
         this.spec = spec;
         this.sharedCsContainer = sharedCsContainer;
         this.clusterName = spec.clusterName();
         if (network != null) {
             this.network = network;
-        } else if (csContainer != null ) {
+        } else if (csContainer != null) {
             this.network = csContainer.getNetwork();
         } else {
             this.network = Network.newNetwork();
@@ -277,7 +270,7 @@ private PulsarCluster(PulsarClusterSpec spec, Network network, CSContainer csCon
                 ));
 
         if (spec.dataContainer != null) {
-            if (csContainer != null) {
+            if (!sharedCsContainer && csContainer != null) {
                 csContainer.withVolumesFrom(spec.dataContainer, BindMode.READ_WRITE);
             }
             if (zkContainer != null) {
@@ -294,7 +287,7 @@ private PulsarCluster(PulsarClusterSpec spec, Network network, CSContainer csCon
             if (zkContainer != null) {
                 zkContainer.withClasspathResourceMapping(key, value, BindMode.READ_WRITE);
             }
-            if (csContainer != null) {
+            if (!sharedCsContainer && csContainer != null) {
                 csContainer.withClasspathResourceMapping(key, value, BindMode.READ_WRITE);
             }
             proxyContainer.withClasspathResourceMapping(key, value, BindMode.READ_WRITE);

tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarClusterSpec.java

@@ -125,7 +125,7 @@ public class PulsarClusterSpec {
     Map<String, String> classPathVolumeMounts = new TreeMap<>();
 
     /**
-     * Daat container
+     * Data container
      */
     @Builder.Default
     GenericContainer<?> dataContainer = null;