pulsar-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From si...@apache.org
Subject [incubator-pulsar] branch master updated: [tests] improve connector related integration tests (#2587)
Date Tue, 18 Sep 2018 21:50:56 GMT
This is an automated email from the ASF dual-hosted git repository.

sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new 7530d64  [tests] improve connector related integration tests (#2587)
7530d64 is described below

commit 7530d64a679a0783122b18058c1148c89c0fee0a
Author: Sijie Guo <guosijie@gmail.com>
AuthorDate: Tue Sep 18 14:50:51 2018 -0700

    [tests] improve connector related integration tests (#2587)
    
    *Motivation*
    
    with more and more connector are added, it becomes expensive to start all external services at the begin.
    
    *Changes*
    
    - refactor the connector testing framework to start external service before methods
    - fix kafka, cassandra and mysql connectors
---
 distribution/io/src/assemble/io.xml                |   6 +
 .../broker/service/PersistentQueueE2ETest.java     |  60 ----------
 .../functions/instance/JavaInstanceRunnable.java   |   3 +-
 .../pulsar/functions/source/TopicSchema.java       |   3 +
 ...rchAbstractSink.java => ElasticSearchSink.java} |   9 +-
 .../io/elasticsearch/ElasticSearchStringSink.java  |  35 ------
 .../resources/META-INF/services/pulsar-io.yaml     |   4 +-
 .../io/elasticsearch/ElasticSearchSinkTests.java   |   4 +-
 .../apache/pulsar/io/kafka/KafkaAbstractSink.java  |   6 +-
 .../{KafkaStringSink.java => KafkaBytesSink.java}  |  25 ++++-
 .../resources/META-INF/services/pulsar-io.yaml     |   2 +-
 site2/docs/io-quickstart.md                        |   2 +-
 .../version-2.1.0-incubating/io-quickstart.md      |   2 +-
 .../integration/functions/PulsarFunctionsTest.java |  39 +++++--
 .../functions/utils/CommandGenerator.java          |   2 +-
 .../integration/io/CassandraSinkArchiveTester.java | 121 ---------------------
 .../tests/integration/io/CassandraSinkTester.java  |  45 +++++---
 .../integration/io/ElasticSearchSinkTester.java    |  34 +++---
 .../tests/integration/io/HdfsSinkTester.java       |  22 ++--
 .../tests/integration/io/JdbcSinkTester.java       |  38 ++++---
 .../tests/integration/io/KafkaSinkTester.java      |  28 +++--
 .../tests/integration/io/KafkaSourceTester.java    |  12 +-
 .../pulsar/tests/integration/io/SinkTester.java    |  25 ++++-
 .../pulsar/tests/integration/io/SourceTester.java  |   4 +-
 .../tests/integration/suites/PulsarTestSuite.java  |  50 ---------
 .../integration/topologies/PulsarCluster.java      |  17 +++
 26 files changed, 212 insertions(+), 386 deletions(-)

diff --git a/distribution/io/src/assemble/io.xml b/distribution/io/src/assemble/io.xml
index 08ff859..a509e19 100644
--- a/distribution/io/src/assemble/io.xml
+++ b/distribution/io/src/assemble/io.xml
@@ -92,5 +92,11 @@
       <outputDirectory>connectors</outputDirectory>
       <fileMode>644</fileMode>
     </file>
+
+    <file>
+      <source>${basedir}/../../pulsar-io/elastic-search/target/pulsar-io-elastic-search-${project.version}.nar</source>
+      <outputDirectory>connectors</outputDirectory>
+      <fileMode>644</fileMode>
+    </file>
   </files>
 </assembly>
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentQueueE2ETest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentQueueE2ETest.java
index 2937ca0..fb7a76c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentQueueE2ETest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentQueueE2ETest.java
@@ -233,66 +233,6 @@ public class PersistentQueueE2ETest extends BrokerTestBase {
         deleteTopic(topicName);
     }
 
-    @Test
-    public void testConsumersWithDifferentPermits() throws Exception {
-        final String topicName = "persistent://prop/use/ns-abc/shared-topic4";
-        final String subName = "sub4";
-        final int numMsgs = 10000;
-
-        final AtomicInteger msgCountConsumer1 = new AtomicInteger(0);
-        final AtomicInteger msgCountConsumer2 = new AtomicInteger(0);
-        final CountDownLatch latch = new CountDownLatch(numMsgs);
-
-        int recvQ1 = 10;
-        Consumer<byte[]> consumer1 = pulsarClient.newConsumer().topic(topicName).subscriptionName(subName)
-                .subscriptionType(SubscriptionType.Shared).receiverQueueSize(recvQ1)
-                .messageListener((consumer, msg) -> {
-                    msgCountConsumer1.incrementAndGet();
-                    try {
-                        consumer.acknowledge(msg);
-                        latch.countDown();
-                    } catch (PulsarClientException e) {
-                        fail("Should not fail");
-                    }
-                }).subscribe();
-
-        int recvQ2 = 1;
-        Consumer<byte[]> consumer2 = pulsarClient.newConsumer().topic(topicName).subscriptionName(subName)
-                .subscriptionType(SubscriptionType.Shared).receiverQueueSize(recvQ2)
-                .messageListener((consumer, msg) -> {
-                    msgCountConsumer2.incrementAndGet();
-                    try {
-                        consumer.acknowledge(msg);
-                        latch.countDown();
-                    } catch (PulsarClientException e) {
-                        fail("Should not fail");
-                    }
-                }).subscribe();
-
-        List<CompletableFuture<MessageId>> futures = Lists.newArrayListWithCapacity(numMsgs);
-        Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName)
-            .enableBatching(false)
-            .maxPendingMessages(numMsgs + 1)
-            .messageRoutingMode(MessageRoutingMode.SinglePartition)
-            .create();
-        for (int i = 0; i < numMsgs; i++) {
-            String message = "msg-" + i;
-            futures.add(producer.sendAsync(message.getBytes()));
-        }
-        FutureUtil.waitForAll(futures).get();
-        producer.close();
-
-        latch.await(5, TimeUnit.SECONDS);
-
-        assertEquals(msgCountConsumer1.get(), numMsgs - numMsgs / (recvQ1 + recvQ2), numMsgs * 0.1);
-        assertEquals(msgCountConsumer2.get(), numMsgs / (recvQ1 + recvQ2), numMsgs * 0.1);
-
-        consumer1.close();
-        consumer2.close();
-
-        deleteTopic(topicName);
-    }
-
     // this test is good to have to see the distribution, but every now and then it gets slightly different than the
     // expected numbers. keeping this disabled to not break the build, but nevertheless this gives good insight into
     // how the round robin distribution algorithm is behaving
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
index 4ba7340..b3f86ea 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
@@ -146,7 +146,8 @@ public class JavaInstanceRunnable implements AutoCloseable, Runnable {
         ThreadContext.put("functionname", instanceConfig.getFunctionDetails().getName());
         ThreadContext.put("instance", instanceConfig.getInstanceId());
 
-        log.info("Starting Java Instance {}", instanceConfig.getFunctionDetails().getName());
+        log.info("Starting Java Instance {} : \n Details = {}",
+            instanceConfig.getFunctionDetails().getName(), instanceConfig.getFunctionDetails());
 
         // start the function thread
         loadJars();
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java
index 76375dc..db9e880 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java
@@ -78,6 +78,9 @@ public class TopicSchema {
     private SchemaType getSchemaTypeOrDefault(String topic, Class<?> clazz) {
         if (GenericRecord.class.isAssignableFrom(clazz)) {
             return SchemaType.AUTO;
+        } else if (byte[].class.equals(clazz)) {
+            // if function uses bytes, we should ignore
+            return SchemaType.NONE;
         } else {
             Optional<SchemaInfo> schema = ((PulsarClientImpl) client).getSchema(topic).join();
             if (schema.isPresent()) {
diff --git a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchAbstractSink.java b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java
similarity index 94%
rename from pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchAbstractSink.java
rename to pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java
index 3760d40..86546f3 100644
--- a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchAbstractSink.java
+++ b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java
@@ -51,7 +51,7 @@ import org.elasticsearch.common.xcontent.XContentType;
  * Users need to implement extractKeyValue function to use this sink.
  * This class assumes that the input will be JSON documents
  */
-public abstract class ElasticSearchAbstractSink<K, V> implements Sink<byte[]> {
+public class ElasticSearchSink implements Sink<byte[]> {
 
     protected static final String DOCUMENT = "doc";
 
@@ -74,7 +74,7 @@ public abstract class ElasticSearchAbstractSink<K, V> implements Sink<byte[]> {
 
     @Override
     public void write(Record<byte[]> record) {
-        KeyValue<K, V> keyValue = extractKeyValue(record);
+        KeyValue<String, byte[]> keyValue = extractKeyValue(record);
         IndexRequest indexRequest = Requests.indexRequest(elasticSearchConfig.getIndexName());
         indexRequest.type(DOCUMENT);
         indexRequest.source(keyValue.getValue(), XContentType.JSON);
@@ -91,7 +91,10 @@ public abstract class ElasticSearchAbstractSink<K, V> implements Sink<byte[]> {
         }
     }
 
-    public abstract KeyValue<K, V> extractKeyValue(Record<byte[]> record);
+    public KeyValue<String, byte[]> extractKeyValue(Record<byte[]> record) {
+        String key = record.getKey().orElseGet(null);
+        return new KeyValue<>(key, record.getValue());
+    }
 
     private void createIndexIfNeeded() throws IOException {
         GetIndexRequest request = new GetIndexRequest();
diff --git a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchStringSink.java b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchStringSink.java
deleted file mode 100644
index 6cfa03d..0000000
--- a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchStringSink.java
+++ /dev/null
@@ -1,35 +0,0 @@
-/**
- * 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.pulsar.io.elasticsearch;
-
-import org.apache.pulsar.functions.api.Record;
-import org.apache.pulsar.io.core.KeyValue;
-
-/**
- * Concrete ElasticSearch sink.
- * This class assumes that the input will be JSON documents
- */
-public class ElasticSearchStringSink extends ElasticSearchAbstractSink<String, String> {
-
-    @Override
-    public KeyValue<String, String> extractKeyValue(Record<byte[]> record) {
-        String key = record.getKey().orElseGet(() -> new String(record.getValue()));
-        return new KeyValue<>(key, new String(record.getValue()));
-    }
-}
diff --git a/pulsar-io/elastic-search/src/main/resources/META-INF/services/pulsar-io.yaml b/pulsar-io/elastic-search/src/main/resources/META-INF/services/pulsar-io.yaml
index 0307516..97789e9 100644
--- a/pulsar-io/elastic-search/src/main/resources/META-INF/services/pulsar-io.yaml
+++ b/pulsar-io/elastic-search/src/main/resources/META-INF/services/pulsar-io.yaml
@@ -17,6 +17,6 @@
 # under the License.
 #
 
-name: Elastic Search
+name: elastic_search
 description: Writes data into Elastic Search
-sinkClass: org.apache.pulsar.io.elasticsearch.ElasticSearchStringSink
+sinkClass: org.apache.pulsar.io.elasticsearch.ElasticSearchSink
diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSinkTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSinkTests.java
index ea2b886..f188829 100644
--- a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSinkTests.java
+++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSinkTests.java
@@ -56,7 +56,7 @@ public class ElasticSearchSinkTests {
     @Mock
     protected SinkContext mockSinkContext;   
     protected Map<String, Object> map;
-    protected ElasticSearchStringSink sink;
+    protected ElasticSearchSink sink;
     
     @BeforeClass
     public static final void init() {
@@ -71,7 +71,7 @@ public class ElasticSearchSinkTests {
     public final void setUp() throws Exception {
         map = new HashMap<String, Object> ();
         map.put("elasticSearchUrl", "http://localhost:9200");
-        sink = new ElasticSearchStringSink();
+        sink = new ElasticSearchSink();
         
         mockRecord = mock(Record.class);
         mockSinkContext = mock(SinkContext.class);
diff --git a/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSink.java b/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSink.java
index a92a368..50ce4b9 100644
--- a/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSink.java
+++ b/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSink.java
@@ -70,6 +70,10 @@ public abstract class KafkaAbstractSink<K, V> implements Sink<byte[]> {
         }
     }
 
+    protected Properties beforeCreateProducer(Properties props) {
+        return props;
+    }
+
     @Override
     public void open(Map<String, Object> config, SinkContext sinkContext) throws Exception {
         kafkaSinkConfig = KafkaSinkConfig.load(config);
@@ -89,7 +93,7 @@ public abstract class KafkaAbstractSink<K, V> implements Sink<byte[]> {
         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, kafkaSinkConfig.getKeySerializerClass());
         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, kafkaSinkConfig.getValueSerializerClass());
 
-        producer = new KafkaProducer<>(props);
+        producer = new KafkaProducer<>(beforeCreateProducer(props));
 
         log.info("Kafka sink started.");
     }
diff --git a/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaStringSink.java b/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaBytesSink.java
similarity index 50%
rename from pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaStringSink.java
rename to pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaBytesSink.java
index 89e3e7f..9ce2bdc 100644
--- a/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaStringSink.java
+++ b/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaBytesSink.java
@@ -19,16 +19,31 @@
 
 package org.apache.pulsar.io.kafka;
 
+import java.util.Properties;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.StringSerializer;
 import org.apache.pulsar.functions.api.Record;
 import org.apache.pulsar.io.core.KeyValue;
 
 /**
- * Kafka sink that treats incoming messages on the input topic as Strings
- * and write identical key/value pairs.
+ * Kafka sink should treats incoming messages as pure bytes. So we don't
+ * apply schema into it.
  */
-public class KafkaStringSink extends KafkaAbstractSink<String, String> {
+@Slf4j
+public class KafkaBytesSink extends KafkaAbstractSink<String, byte[]> {
+
+    @Override
+    protected Properties beforeCreateProducer(Properties props) {
+        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+        log.info("Created kafka producer config : {}", props);
+        return props;
+    }
+
     @Override
-    public KeyValue<String, String> extractKeyValue(Record<byte[]> record) {
-        return new KeyValue<>(record.getKey().orElse(null), new String(record.getValue()));
+    public KeyValue<String, byte[]> extractKeyValue(Record<byte[]> record) {
+        return new KeyValue<>(record.getKey().orElse(null), record.getValue());
     }
 }
\ No newline at end of file
diff --git a/pulsar-io/kafka/src/main/resources/META-INF/services/pulsar-io.yaml b/pulsar-io/kafka/src/main/resources/META-INF/services/pulsar-io.yaml
index a7bc813..7afc154 100644
--- a/pulsar-io/kafka/src/main/resources/META-INF/services/pulsar-io.yaml
+++ b/pulsar-io/kafka/src/main/resources/META-INF/services/pulsar-io.yaml
@@ -20,4 +20,4 @@
 name: kafka
 description: Kafka source and sink connector
 sourceClass: org.apache.pulsar.io.kafka.KafkaStringSource
-sinkClass: org.apache.pulsar.io.kafka.KafkaStringSink
+sinkClass: org.apache.pulsar.io.kafka.KafkaBytesSink
diff --git a/site2/docs/io-quickstart.md b/site2/docs/io-quickstart.md
index 8b8cfd3..4a48bc0 100644
--- a/site2/docs/io-quickstart.md
+++ b/site2/docs/io-quickstart.md
@@ -123,7 +123,7 @@ curl -s http://localhost:8080/admin/v2/functions/connectors
 
 Example output:
 ```json
-[{"name":"aerospike","description":"Aerospike database sink","sinkClass":"org.apache.pulsar.io.aerospike.AerospikeStringSink"},{"name":"cassandra","description":"Writes data into Cassandra","sinkClass":"org.apache.pulsar.io.cassandra.CassandraStringSink"},{"name":"kafka","description":"Kafka source and sink connector","sourceClass":"org.apache.pulsar.io.kafka.KafkaStringSource","sinkClass":"org.apache.pulsar.io.kafka.KafkaStringSink"},{"name":"kinesis","description":"Kinesis sink connect [...]
+[{"name":"aerospike","description":"Aerospike database sink","sinkClass":"org.apache.pulsar.io.aerospike.AerospikeStringSink"},{"name":"cassandra","description":"Writes data into Cassandra","sinkClass":"org.apache.pulsar.io.cassandra.CassandraStringSink"},{"name":"kafka","description":"Kafka source and sink connector","sourceClass":"org.apache.pulsar.io.kafka.KafkaStringSource","sinkClass":"org.apache.pulsar.io.kafka.KafkaBytesSink"},{"name":"kinesis","description":"Kinesis sink connecto [...]
 ```
 
 If an error occurred while starting Pulsar service, you may be able to seen exception at the terminal you are running `pulsar/standalone`,
diff --git a/site2/website/versioned_docs/version-2.1.0-incubating/io-quickstart.md b/site2/website/versioned_docs/version-2.1.0-incubating/io-quickstart.md
index 7b21379..afa8e31 100644
--- a/site2/website/versioned_docs/version-2.1.0-incubating/io-quickstart.md
+++ b/site2/website/versioned_docs/version-2.1.0-incubating/io-quickstart.md
@@ -124,7 +124,7 @@ curl -s http://localhost:8080/admin/v2/functions/connectors
 
 Example output:
 ```json
-[{"name":"aerospike","description":"Aerospike database sink","sinkClass":"org.apache.pulsar.io.aerospike.AerospikeStringSink"},{"name":"cassandra","description":"Writes data into Cassandra","sinkClass":"org.apache.pulsar.io.cassandra.CassandraStringSink"},{"name":"kafka","description":"Kafka source and sink connector","sourceClass":"org.apache.pulsar.io.kafka.KafkaStringSource","sinkClass":"org.apache.pulsar.io.kafka.KafkaStringSink"},{"name":"kinesis","description":"Kinesis sink connect [...]
+[{"name":"aerospike","description":"Aerospike database sink","sinkClass":"org.apache.pulsar.io.aerospike.AerospikeStringSink"},{"name":"cassandra","description":"Writes data into Cassandra","sinkClass":"org.apache.pulsar.io.cassandra.CassandraStringSink"},{"name":"kafka","description":"Kafka source and sink connector","sourceClass":"org.apache.pulsar.io.kafka.KafkaStringSource","sinkClass":"org.apache.pulsar.io.kafka.KafkaBytesSink"},{"name":"kinesis","description":"Kinesis sink connecto [...]
 ```
 
 If an error occurred while starting Pulsar service, you may be able to seen exception at the terminal you are running `pulsar/standalone`,
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java
index 5f11eef..e9ea618 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java
@@ -48,6 +48,7 @@ import org.apache.pulsar.tests.integration.io.*;
 import org.apache.pulsar.tests.integration.io.JdbcSinkTester.Foo;
 import org.apache.pulsar.tests.integration.topologies.FunctionRuntimeType;
 import org.apache.pulsar.tests.integration.topologies.PulsarCluster;
+import org.testcontainers.containers.GenericContainer;
 import org.testng.annotations.Test;
 
 /**
@@ -62,17 +63,17 @@ public abstract class PulsarFunctionsTest extends PulsarFunctionsTestBase {
 
     @Test
     public void testKafkaSink() throws Exception {
-        testSink(new KafkaSinkTester(), true);
+        testSink(new KafkaSinkTester(), true, new KafkaSourceTester());
     }
 
     @Test
     public void testCassandraSink() throws Exception {
-        testSink(new CassandraSinkTester(), true);
+        testSink(CassandraSinkTester.createTester(true), true);
     }
 
     @Test
     public void testCassandraArchiveSink() throws Exception {
-        testSink(new CassandraSinkArchiveTester(), false);
+        testSink(CassandraSinkTester.createTester(false), false);
     }
     
     @Test(enabled = false)
@@ -91,8 +92,31 @@ public abstract class PulsarFunctionsTest extends PulsarFunctionsTestBase {
     }
     
     private void testSink(SinkTester tester, boolean builtin) throws Exception {
-        tester.findSinkServiceContainer(pulsarCluster.getExternalServices());
+        tester.startServiceContainer(pulsarCluster);
+        try {
+            runSinkTester(tester, builtin);
+        } finally {
+            tester.stopServiceContainer(pulsarCluster);
+        }
+    }
+
 
+    private <ServiceContainerT extends GenericContainer>  void testSink(SinkTester<ServiceContainerT> sinkTester,
+                                                                        boolean builtinSink,
+                                                                        SourceTester<ServiceContainerT> sourceTester)
+            throws Exception {
+        ServiceContainerT serviceContainer = sinkTester.startServiceContainer(pulsarCluster);
+        try {
+            runSinkTester(sinkTester, builtinSink);
+            if (null != sourceTester) {
+                sourceTester.setServiceContainer(serviceContainer);
+                testSource(sourceTester);
+            }
+        } finally {
+            sinkTester.stopServiceContainer(pulsarCluster);
+        }
+    }
+    private void runSinkTester(SinkTester tester, boolean builtin) throws Exception {
         final String tenant = TopicName.PUBLIC_TENANT;
         final String namespace = TopicName.DEFAULT_NAMESPACE;
         final String inputTopicName = "test-sink-connector-"
@@ -357,14 +381,7 @@ public abstract class PulsarFunctionsTest extends PulsarFunctionsTestBase {
     // Source Test
     //
 
-    @Test
-    public void testKafkaSource() throws Exception {
-        testSource(new KafkaSourceTester());
-    }
-
     private void testSource(SourceTester tester)  throws Exception {
-        tester.findSourceServiceContainer(pulsarCluster.getExternalServices());
-
         final String tenant = TopicName.PUBLIC_TENANT;
         final String namespace = TopicName.DEFAULT_NAMESPACE;
         final String outputTopicName = "test-source-connector-"
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/utils/CommandGenerator.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/utils/CommandGenerator.java
index 6578d08..6f4d012 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/utils/CommandGenerator.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/utils/CommandGenerator.java
@@ -161,7 +161,7 @@ public class CommandGenerator {
     }
 
     public String generateUpdateFunctionCommand(String codeFile) {
-        StringBuilder commandBuilder = new StringBuilder("PULSAR_MEM=-Xmx1024m ");
+        StringBuilder commandBuilder = new StringBuilder();
         if (adminUrl == null) {
             commandBuilder.append("/pulsar/bin/pulsar-admin functions update");
         } else {
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/CassandraSinkArchiveTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/CassandraSinkArchiveTester.java
deleted file mode 100644
index 86c7689..0000000
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/CassandraSinkArchiveTester.java
+++ /dev/null
@@ -1,121 +0,0 @@
-/**
- * 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.pulsar.tests.integration.io;
-
-import com.datastax.driver.core.Cluster;
-import com.datastax.driver.core.ResultSet;
-import com.datastax.driver.core.Row;
-import com.datastax.driver.core.Session;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.pulsar.tests.integration.containers.CassandraContainer;
-import org.testcontainers.containers.GenericContainer;
-
-import java.util.List;
-import java.util.Map;
-
-import static com.google.common.base.Preconditions.checkState;
-import static org.apache.pulsar.tests.integration.topologies.PulsarClusterTestBase.randomName;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-
-/**
- * A tester for testing cassandra sink submitted as an archive.
- */
-@Slf4j
-public class CassandraSinkArchiveTester extends SinkTester {
-
-    private static final String NAME = "cassandra";
-
-    private static final String ROOTS = "cassandra";
-    private static final String KEY = "key";
-    private static final String COLUMN = "col";
-
-    private final String keySpace;
-    private final String tableName;
-
-    private CassandraContainer cassandraCluster;
-
-    private Cluster cluster;
-    private Session session;
-
-    public CassandraSinkArchiveTester() {
-        super("/pulsar/connectors/pulsar-io-cassandra-2.2.0-incubating-SNAPSHOT.nar", "org.apache.pulsar.io.cassandra.CassandraStringSink");
-
-        String suffix = randomName(8) + "_" + System.currentTimeMillis();
-        this.keySpace = "keySpace_" + suffix;
-        this.tableName = "tableName_" + suffix;
-
-        sinkConfig.put("roots", ROOTS);
-        sinkConfig.put("keyspace", keySpace);
-        sinkConfig.put("columnFamily", tableName);
-        sinkConfig.put("keyname", KEY);
-        sinkConfig.put("columnName", COLUMN);
-    }
-
-    @Override
-    public void findSinkServiceContainer(Map<String, GenericContainer<?>> containers) {
-        GenericContainer<?> container = containers.get(NAME);
-        checkState(container instanceof CassandraContainer,
-            "No kafka service found in the cluster");
-
-        this.cassandraCluster = (CassandraContainer) container;
-    }
-
-    @Override
-    public void prepareSink() {
-        // build the sink
-        cluster = Cluster.builder()
-            .addContactPoint("localhost")
-            .withPort(cassandraCluster.getCassandraPort())
-            .build();
-
-        // connect to the cluster
-        session = cluster.connect();
-        log.info("Connecting to cassandra cluster at localhost:{}", cassandraCluster.getCassandraPort());
-
-        String createKeySpace =
-            "CREATE KEYSPACE " + keySpace
-                + " WITH replication = {'class':'SimpleStrategy', 'replication_factor':1}; ";
-        log.info(createKeySpace);
-        session.execute(createKeySpace);
-        session.execute("USE " + keySpace);
-
-        String createTable = "CREATE TABLE " + tableName
-            + "(" + KEY + " text PRIMARY KEY, "
-            + COLUMN + " text);";
-        log.info(createTable);
-        session.execute(createTable);
-    }
-
-    @Override
-    public void validateSinkResult(Map<String, String> kvs) {
-        String query = "SELECT * FROM " + tableName + ";";
-        ResultSet result = session.execute(query);
-        List<Row> rows = result.all();
-        assertEquals(kvs.size(), rows.size());
-        for (Row row : rows) {
-            String key = row.getString(KEY);
-            String value = row.getString(COLUMN);
-
-            String expectedValue = kvs.get(key);
-            assertNotNull(expectedValue);
-            assertEquals(expectedValue, value);
-        }
-    }
-}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/CassandraSinkTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/CassandraSinkTester.java
index c9d3e5a..3309358 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/CassandraSinkTester.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/CassandraSinkTester.java
@@ -24,12 +24,11 @@ import com.datastax.driver.core.Row;
 import com.datastax.driver.core.Session;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.tests.integration.containers.CassandraContainer;
-import org.testcontainers.containers.GenericContainer;
+import org.apache.pulsar.tests.integration.topologies.PulsarCluster;
 
 import java.util.List;
 import java.util.Map;
 
-import static com.google.common.base.Preconditions.checkState;
 import static org.apache.pulsar.tests.integration.topologies.PulsarClusterTestBase.randomName;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
@@ -38,7 +37,15 @@ import static org.junit.Assert.assertNotNull;
  * A tester for testing cassandra sink.
  */
 @Slf4j
-public class CassandraSinkTester extends SinkTester {
+public class CassandraSinkTester extends SinkTester<CassandraContainer> {
+
+    public static CassandraSinkTester createTester(boolean builtin) {
+        if (builtin) {
+            return new CassandraSinkTester(builtin);
+        } else {
+            return new CassandraSinkTester();
+        }
+    }
 
     private static final String NAME = "cassandra";
 
@@ -49,13 +56,11 @@ public class CassandraSinkTester extends SinkTester {
     private final String keySpace;
     private final String tableName;
 
-    private CassandraContainer cassandraCluster;
-
     private Cluster cluster;
     private Session session;
 
-    public CassandraSinkTester() {
-        super(SinkType.CASSANDRA);
+    private CassandraSinkTester() {
+        super(NAME, "/pulsar/connectors/pulsar-io-cassandra-2.2.0-incubating-SNAPSHOT.nar", "org.apache.pulsar.io.cassandra.CassandraStringSink");
 
         String suffix = randomName(8) + "_" + System.currentTimeMillis();
         this.keySpace = "keySpace_" + suffix;
@@ -68,13 +73,23 @@ public class CassandraSinkTester extends SinkTester {
         sinkConfig.put("columnName", COLUMN);
     }
 
-    @Override
-    public void findSinkServiceContainer(Map<String, GenericContainer<?>> containers) {
-        GenericContainer<?> container = containers.get(NAME);
-        checkState(container instanceof CassandraContainer,
-            "No kafka service found in the cluster");
+    private CassandraSinkTester(boolean builtin) {
+        super(NAME, SinkType.CASSANDRA);
+
+        String suffix = randomName(8) + "_" + System.currentTimeMillis();
+        this.keySpace = "keySpace_" + suffix;
+        this.tableName = "tableName_" + suffix;
 
-        this.cassandraCluster = (CassandraContainer) container;
+        sinkConfig.put("roots", ROOTS);
+        sinkConfig.put("keyspace", keySpace);
+        sinkConfig.put("columnFamily", tableName);
+        sinkConfig.put("keyname", KEY);
+        sinkConfig.put("columnName", COLUMN);
+    }
+
+    @Override
+    protected CassandraContainer createSinkService(PulsarCluster cluster) {
+        return new CassandraContainer(cluster.getClusterName());
     }
 
     @Override
@@ -82,12 +97,12 @@ public class CassandraSinkTester extends SinkTester {
         // build the sink
         cluster = Cluster.builder()
             .addContactPoint("localhost")
-            .withPort(cassandraCluster.getCassandraPort())
+            .withPort(serviceContainer.getCassandraPort())
             .build();
 
         // connect to the cluster
         session = cluster.connect();
-        log.info("Connecting to cassandra cluster at localhost:{}", cassandraCluster.getCassandraPort());
+        log.info("Connecting to cassandra cluster at localhost:{}", serviceContainer.getCassandraPort());
 
         String createKeySpace =
             "CREATE KEYSPACE " + keySpace
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/ElasticSearchSinkTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/ElasticSearchSinkTester.java
index 0effc8e..eee208e 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/ElasticSearchSinkTester.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/ElasticSearchSinkTester.java
@@ -18,57 +18,57 @@
  */
 package org.apache.pulsar.tests.integration.io;
 
-import static com.google.common.base.Preconditions.checkState;
 import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
 
 import java.util.Map;
 
-import org.apache.http.Header;
 import org.apache.http.HttpHost;
 import org.apache.pulsar.tests.integration.containers.ElasticSearchContainer;
+import org.apache.pulsar.tests.integration.topologies.PulsarCluster;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.client.RestClient;
 import org.elasticsearch.client.RestClientBuilder;
 import org.elasticsearch.client.RestHighLevelClient;
-import org.testcontainers.containers.GenericContainer;
 
-public class ElasticSearchSinkTester extends SinkTester {
-    
+public class ElasticSearchSinkTester extends SinkTester<ElasticSearchContainer> {
+
     private RestHighLevelClient elasticClient;
 
     public ElasticSearchSinkTester() {
-        super(SinkType.ELASTIC_SEARCH);
+        super(ElasticSearchContainer.NAME, SinkType.ELASTIC_SEARCH);
         
-        sinkConfig.put("elasticSearchUrl", "http://localhost:9200");
+        sinkConfig.put("elasticSearchUrl", "http://" + ElasticSearchContainer.NAME + ":9200");
         sinkConfig.put("indexName", "test-index");
     }
 
+
     @Override
-    public void findSinkServiceContainer(Map<String, GenericContainer<?>> externalServices) {
-        GenericContainer<?> container = externalServices.get(ElasticSearchContainer.NAME);
-        checkState(container instanceof ElasticSearchContainer,
-            "No ElasticSearch service found in the cluster");
+    protected ElasticSearchContainer createSinkService(PulsarCluster cluster) {
+        return new ElasticSearchContainer(cluster.getClusterName());
     }
 
     @Override
     public void prepareSink() throws Exception {
-        RestClientBuilder builder = RestClient.builder(new HttpHost("localhost", 9200, "http"));
+        RestClientBuilder builder = RestClient.builder(
+            new HttpHost(
+                "localhost",
+                serviceContainer.getMappedPort(9200),
+                "http"));
         elasticClient = new RestHighLevelClient(builder);
     }
 
     @Override
     public void validateSinkResult(Map<String, String> kvs) {
-        
         SearchRequest searchRequest = new SearchRequest("test-index");
         searchRequest.types("doc");
         
         try {
-            Header headers = null;
-            SearchResponse searchResult = elasticClient.search(searchRequest, headers);
-            assertTrue(searchResult.getHits().getTotalHits() > 0);
+            SearchResponse searchResult = elasticClient.search(searchRequest);
+            assertTrue(searchResult.getHits().getTotalHits() > 0, searchResult.toString());
         } catch (Exception e) {
-            e.printStackTrace();
+            fail("Encountered exception on validating elastic search results", e);
         }
     }
 
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/HdfsSinkTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/HdfsSinkTester.java
index 46c5f24..957b93a 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/HdfsSinkTester.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/HdfsSinkTester.java
@@ -21,18 +21,14 @@ package org.apache.pulsar.tests.integration.io;
 import java.util.Map;
 
 import org.apache.pulsar.tests.integration.containers.HdfsContainer;
-import org.testcontainers.containers.GenericContainer;
+import org.apache.pulsar.tests.integration.topologies.PulsarCluster;
 
-import static com.google.common.base.Preconditions.checkState;
-
-public class HdfsSinkTester extends SinkTester {
+public class HdfsSinkTester extends SinkTester<HdfsContainer> {
 	
 	private static final String NAME = "HDFS";
 	
-	private HdfsContainer hdfsCluster;
-	
 	public HdfsSinkTester() {
-		super(SinkType.HDFS);
+		super(NAME, SinkType.HDFS);
 		
 		// TODO How do I get the core-site.xml, and hdfs-site.xml files from the container?
 		sinkConfig.put("hdfsConfigResources", "");
@@ -40,20 +36,18 @@ public class HdfsSinkTester extends SinkTester {
 	}
 
 	@Override
-	public void findSinkServiceContainer(Map<String, GenericContainer<?>> containers) {
-		GenericContainer<?> container = containers.get(NAME);	
-		checkState(container instanceof HdfsContainer, "No HDFS service found in the cluster");
-	    this.hdfsCluster = (HdfsContainer) container;
+	protected HdfsContainer createSinkService(PulsarCluster cluster) {
+		return new HdfsContainer(cluster.getClusterName());
 	}
 
 	@Override
 	public void prepareSink() throws Exception {
 		// Create the test directory
-		hdfsCluster.execInContainer("/hadoop/bin/hdfs","dfs", "-mkdir", "/tmp/testing");
-		hdfsCluster.execInContainer("/hadoop/bin/hdfs", "-chown", "tester:testing", "/tmp/testing");
+		serviceContainer.execInContainer("/hadoop/bin/hdfs","dfs", "-mkdir", "/tmp/testing");
+		serviceContainer.execInContainer("/hadoop/bin/hdfs", "-chown", "tester:testing", "/tmp/testing");
 		
 		// Execute all future commands as the "tester" user
-		hdfsCluster.execInContainer("export HADOOP_USER_NAME=tester");
+		serviceContainer.execInContainer("export HADOOP_USER_NAME=tester");
 	}
 
 	@Override
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/JdbcSinkTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/JdbcSinkTester.java
index 7c14ba9..72d9b01 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/JdbcSinkTester.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/JdbcSinkTester.java
@@ -18,22 +18,23 @@
  */
 package org.apache.pulsar.tests.integration.io;
 
-import static com.google.common.base.Preconditions.checkState;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.fail;
 
+import com.github.dockerjava.api.command.CreateContainerCmd;
 import java.sql.Connection;
 import java.sql.DriverManager;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.util.Map;
+import java.util.function.Consumer;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
 import lombok.ToString;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.impl.schema.AvroSchema;
-import org.testcontainers.containers.GenericContainer;
+import org.apache.pulsar.tests.integration.topologies.PulsarCluster;
 import org.testcontainers.containers.MySQLContainer;
 
 /**
@@ -41,7 +42,7 @@ import org.testcontainers.containers.MySQLContainer;
  * This will use MySql as DB server
  */
 @Slf4j
-public class JdbcSinkTester extends SinkTester {
+public class JdbcSinkTester extends SinkTester<MySQLContainer> {
 
     /**
      * A Simple class to test jdbc class´╝î
@@ -57,14 +58,14 @@ public class JdbcSinkTester extends SinkTester {
     }
 
     private static final String NAME = "jdbc";
+    private static final String MYSQL = "mysql";
 
-    private MySQLContainer mySQLContainer;
     private AvroSchema<Foo> schema = AvroSchema.of(Foo.class);
     private String tableName = "test";
     private Connection connection;
 
     public JdbcSinkTester() {
-        super(SinkType.JDBC);
+        super(NAME, SinkType.JDBC);
 
         // container default value is test
         sinkConfig.put("userName", "test");
@@ -79,21 +80,28 @@ public class JdbcSinkTester extends SinkTester {
     }
 
     @Override
-    public void findSinkServiceContainer(Map<String, GenericContainer<?>> containers) {
-        GenericContainer<?> container = containers.get("mysql");
-        checkState(container instanceof MySQLContainer,
-            "No MySQL service found in the cluster");
-
-        this.mySQLContainer = (MySQLContainer) container;
-        log.info("find sink service container: {}", mySQLContainer.getContainerName());
+    protected MySQLContainer createSinkService(PulsarCluster cluster) {
+        return (MySQLContainer) new MySQLContainer()
+            .withUsername("test")
+            .withPassword("test")
+            .withDatabaseName("test")
+            .withNetworkAliases(MYSQL)
+            .withCreateContainerCmdModifier(new Consumer<CreateContainerCmd>() {
+                @Override
+                public void accept(CreateContainerCmd createContainerCmd) {
+                    createContainerCmd
+                        .withName(MYSQL)
+                        .withHostName(cluster.getClusterName() + "-" + MYSQL);
+                }
+            });
     }
 
     @Override
     public void prepareSink() throws Exception {
-        String jdbcUrl = mySQLContainer.getJdbcUrl();
+        String jdbcUrl = serviceContainer.getJdbcUrl();
         // we need set mysql server address in cluster network.
-        sinkConfig.put("jdbcUrl", "jdbc:mysql://mysql:3306/test");
-        String driver = mySQLContainer.getDriverClassName();
+        sinkConfig.put("jdbcUrl", "jdbc:mysql://" + MYSQL + ":3306/test");
+        String driver = serviceContainer.getDriverClassName();
         Class.forName(driver);
 
         connection = DriverManager.getConnection(jdbcUrl, "test", "test");
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/KafkaSinkTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/KafkaSinkTester.java
index ff79e1a..6713cc1 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/KafkaSinkTester.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/KafkaSinkTester.java
@@ -18,7 +18,6 @@
  */
 package org.apache.pulsar.tests.integration.io;
 
-import static com.google.common.base.Preconditions.checkState;
 import static org.apache.pulsar.tests.integration.topologies.PulsarClusterTestBase.randomName;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertTrue;
@@ -32,8 +31,8 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.pulsar.tests.integration.topologies.PulsarCluster;
 import org.testcontainers.containers.Container.ExecResult;
-import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.containers.KafkaContainer;
 import org.testcontainers.shaded.com.google.common.collect.ImmutableMap;
 
@@ -41,18 +40,15 @@ import org.testcontainers.shaded.com.google.common.collect.ImmutableMap;
  * A tester for testing kafka sink.
  */
 @Slf4j
-public class KafkaSinkTester extends SinkTester {
+public class KafkaSinkTester extends SinkTester<KafkaContainer> {
 
     private static final String NAME = "kafka";
 
     private final String kafkaTopicName;
-
-    private KafkaContainer kafkaContainer;
-
     private KafkaConsumer<String, String> kafkaConsumer;
 
     public KafkaSinkTester() {
-        super(SinkType.KAFKA);
+        super(NAME, SinkType.KAFKA);
         String suffix = randomName(8) + "_" + System.currentTimeMillis();
         this.kafkaTopicName = "kafka_sink_topic_" + suffix;
 
@@ -64,17 +60,19 @@ public class KafkaSinkTester extends SinkTester {
     }
 
     @Override
-    public void findSinkServiceContainer(Map<String, GenericContainer<?>> containers) {
-        GenericContainer<?> container = containers.get(NAME);
-        checkState(container instanceof KafkaContainer,
-            "No kafka service found in the cluster");
-
-        this.kafkaContainer = (KafkaContainer) container;
+    protected KafkaContainer createSinkService(PulsarCluster cluster) {
+        final String kafkaServiceName = NAME;
+        return new KafkaContainer()
+                .withEmbeddedZookeeper()
+                .withNetworkAliases(kafkaServiceName)
+                .withCreateContainerCmdModifier(createContainerCmd -> createContainerCmd
+                    .withName(kafkaServiceName)
+                    .withHostName(cluster.getClusterName() + "-" + kafkaServiceName));
     }
 
     @Override
     public void prepareSink() throws Exception {
-        ExecResult execResult = kafkaContainer.execInContainer(
+        ExecResult execResult = serviceContainer.execInContainer(
             "/usr/bin/kafka-topics",
             "--create",
             "--zookeeper",
@@ -91,7 +89,7 @@ public class KafkaSinkTester extends SinkTester {
 
         kafkaConsumer = new KafkaConsumer<>(
             ImmutableMap.of(
-                ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainer.getBootstrapServers(),
+                ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, serviceContainer.getBootstrapServers(),
                 ConsumerConfig.GROUP_ID_CONFIG, "sink-test-" + randomName(8),
                 ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"
             ),
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/KafkaSourceTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/KafkaSourceTester.java
index 4928f00..cee17b1 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/KafkaSourceTester.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/KafkaSourceTester.java
@@ -18,7 +18,6 @@
  */
 package org.apache.pulsar.tests.integration.io;
 
-import static com.google.common.base.Preconditions.checkState;
 import static org.apache.pulsar.tests.integration.topologies.PulsarClusterTestBase.randomName;
 import static org.testng.Assert.assertTrue;
 
@@ -35,7 +34,6 @@ import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.testcontainers.containers.Container.ExecResult;
-import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.containers.KafkaContainer;
 import org.testcontainers.shaded.com.google.common.collect.ImmutableMap;
 
@@ -43,7 +41,7 @@ import org.testcontainers.shaded.com.google.common.collect.ImmutableMap;
  * A tester for testing kafka source.
  */
 @Slf4j
-public class KafkaSourceTester extends SourceTester {
+public class KafkaSourceTester extends SourceTester<KafkaContainer> {
 
     private static final String NAME = "kafka";
 
@@ -68,12 +66,8 @@ public class KafkaSourceTester extends SourceTester {
     }
 
     @Override
-    public void findSourceServiceContainer(Map<String, GenericContainer<?>> containers) {
-        GenericContainer<?> container = containers.get(NAME);
-        checkState(container instanceof KafkaContainer,
-            "No kafka service found in the cluster");
-
-        this.kafkaContainer = (KafkaContainer) container;
+    public void setServiceContainer(KafkaContainer container) {
+        this.kafkaContainer = container;
     }
 
     @Override
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/SinkTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/SinkTester.java
index d2917e6..2dd0759 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/SinkTester.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/SinkTester.java
@@ -21,6 +21,7 @@ package org.apache.pulsar.tests.integration.io;
 import java.util.Map;
 import lombok.Getter;
 import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.tests.integration.topologies.PulsarCluster;
 import org.testcontainers.containers.GenericContainer;
 import org.testng.collections.Maps;
 
@@ -28,7 +29,7 @@ import org.testng.collections.Maps;
  * A tester used for testing a specific sink.
  */
 @Getter
-public abstract class SinkTester {
+public abstract class SinkTester<ServiceContainerT extends GenericContainer> {
 
     public enum SinkType {
         UNDEFINED,
@@ -39,19 +40,23 @@ public abstract class SinkTester {
         ELASTIC_SEARCH
     }
 
+    protected final String networkAlias;
     protected final SinkType sinkType;
     protected final String sinkArchive;
     protected final String sinkClassName;
     protected final Map<String, Object> sinkConfig;
+    protected ServiceContainerT serviceContainer;
 
-    public SinkTester(SinkType sinkType) {
+    public SinkTester(String networkAlias, SinkType sinkType) {
+        this.networkAlias = networkAlias;
         this.sinkType = sinkType;
         this.sinkArchive = null;
         this.sinkClassName = null;
         this.sinkConfig = Maps.newHashMap();
     }
 
-    public SinkTester(String sinkArchive, String sinkClassName) {
+    public SinkTester(String networkAlias, String sinkArchive, String sinkClassName) {
+        this.networkAlias = networkAlias;
         this.sinkType = SinkType.UNDEFINED;
         this.sinkArchive = sinkArchive;
         this.sinkClassName = sinkClassName;
@@ -62,7 +67,19 @@ public abstract class SinkTester {
         return Schema.STRING;
     }
 
-    public abstract void findSinkServiceContainer(Map<String, GenericContainer<?>> externalServices);
+    protected abstract ServiceContainerT createSinkService(PulsarCluster cluster);
+
+    public ServiceContainerT startServiceContainer(PulsarCluster cluster) {
+        this.serviceContainer = createSinkService(cluster);
+        cluster.startService(networkAlias, serviceContainer);
+        return serviceContainer;
+    }
+
+    public void stopServiceContainer(PulsarCluster cluster) {
+        if (null != serviceContainer) {
+            cluster.stopService(networkAlias, serviceContainer);
+        }
+    }
 
     public SinkType sinkType() {
         return sinkType;
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/SourceTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/SourceTester.java
index dc58f2f..f1feb70 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/SourceTester.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/SourceTester.java
@@ -27,7 +27,7 @@ import org.testng.collections.Maps;
  * A tester used for testing a specific source.
  */
 @Getter
-public abstract class SourceTester {
+public abstract class SourceTester<ServiceContainerT extends GenericContainer> {
 
     protected final String sourceType;
     protected final Map<String, Object> sourceConfig;
@@ -37,7 +37,7 @@ public abstract class SourceTester {
         this.sourceConfig = Maps.newHashMap();
     }
 
-    public abstract void findSourceServiceContainer(Map<String, GenericContainer<?>> externalServices);
+    public abstract void setServiceContainer(ServiceContainerT serviceContainer);
 
     public String sourceType() {
         return sourceType;
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/suites/PulsarTestSuite.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/suites/PulsarTestSuite.java
index b4a6b83..20b9da0 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/suites/PulsarTestSuite.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/suites/PulsarTestSuite.java
@@ -18,19 +18,10 @@
  */
 package org.apache.pulsar.tests.integration.suites;
 
-import java.util.Map;
-import org.apache.pulsar.tests.integration.containers.CassandraContainer;
-import org.apache.pulsar.tests.integration.containers.ElasticSearchContainer;
-import org.apache.pulsar.tests.integration.containers.HdfsContainer;
-import org.apache.pulsar.tests.integration.topologies.PulsarClusterSpec.PulsarClusterSpecBuilder;
 import org.apache.pulsar.tests.integration.topologies.PulsarClusterTestBase;
-import org.testcontainers.containers.GenericContainer;
-import org.testcontainers.containers.KafkaContainer;
-import org.testcontainers.containers.MySQLContainer;
 import org.testng.ITest;
 import org.testng.annotations.AfterSuite;
 import org.testng.annotations.BeforeSuite;
-import org.testng.collections.Maps;
 
 public class PulsarTestSuite extends PulsarClusterTestBase implements ITest {
 
@@ -47,47 +38,6 @@ public class PulsarTestSuite extends PulsarClusterTestBase implements ITest {
     }
 
     @Override
-    protected PulsarClusterSpecBuilder beforeSetupCluster(String clusterName, PulsarClusterSpecBuilder specBuilder) {
-        PulsarClusterSpecBuilder builder = super.beforeSetupCluster(clusterName, specBuilder);
-
-        // start functions
-
-        // register external services
-        Map<String, GenericContainer<?>> externalServices = Maps.newHashMap();
-
-        final String kafkaServiceName = "kafka";
-        externalServices.put(
-            kafkaServiceName,
-            new KafkaContainer()
-                .withEmbeddedZookeeper()
-                .withNetworkAliases(kafkaServiceName)
-                .withCreateContainerCmdModifier(createContainerCmd -> createContainerCmd
-                    .withName(kafkaServiceName)
-                    .withHostName(clusterName + "-" + kafkaServiceName)));
-
-        final String cassandraServiceName = "cassandra";
-        externalServices.put(
-            cassandraServiceName,
-            new CassandraContainer(clusterName));
-
-        // use mySQL for jdbc test
-        final String jdbcServiceName = "mysql";
-        externalServices.put(
-            jdbcServiceName,
-            new MySQLContainer()
-                .withExposedPorts(3306));
-        
-        externalServices.put(
-                ElasticSearchContainer.NAME, 
-                new ElasticSearchContainer(ElasticSearchContainer.NAME)
-                .withExposedPorts(9200));
-
-        builder = builder.externalServices(externalServices);
-
-        return builder;
-    }
-
-    @Override
     public String getTestName() {
         return "pulsar-test-suite";
     }
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java
index f78097b..af60712 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java
@@ -209,6 +209,23 @@ public class PulsarCluster {
         }
     }
 
+    public void startService(String networkAlias,
+                             GenericContainer<?> serviceContainer) {
+        log.info("Starting external service {} ...", networkAlias);
+        serviceContainer.withNetwork(network);
+        serviceContainer.withNetworkAliases(networkAlias);
+        serviceContainer.start();
+        log.info("Successfully start external service {}", networkAlias);
+    }
+
+    public void stopService(String networkAlias,
+                            GenericContainer<?> serviceContainer) {
+        log.info("Stopping external service {} ...", networkAlias);
+        serviceContainer.stop();
+        log.info("Successfully stop external service {}", networkAlias);
+    }
+
+
     private static <T extends PulsarContainer> Map<String, T> runNumContainers(String serviceName,
                                                                                int numContainers,
                                                                                Function<String, T> containerCreator) {


Mime
View raw message