[Github actions]Fix integration test disk space is full (#6098)
### Motivation
At present, there are about 28 GB of space available on the running machine of Github action, and there is less than 10GB of disk space left after packing with the `mvn install -DskipTests -Pdocker` command. therefore, the bookie is very easy to exit due to insufficient disk in the integration test. therefore, fix this problem and pack only the needed images.
The current integration tests `PulsarFunctionsTest` are all in one class, which is very inconvenient to manage. For example, after the relevant tests of debezium are run, we can clean up the unused images of debezium, so adding groups to the integration test cases, so that we can control the tests more conveniently in the future.
Now we have about 20G of space for testing.
```
df -h
Filesystem Size Used Avail Use% Mounted on
udev 3.4G 0 3.4G 0% /dev
tmpfs 695M 688K 694M 1% /run
/dev/sda1 84G 64G 20G 77% /
tmpfs 3.4G 8.0K 3.4G 1% /dev/shm
tmpfs 5.0M 0 5.0M 0% /run/lock
tmpfs 3.4G 0 3.4G 0% /sys/fs/cgroup
/dev/sda15 105M 3.6M 101M 4% /boot/efi
/dev/sdb1 14G 41M 13G 1% /mnt
```
```
docker images
REPOSITORY TAG IMAGE ID CREATED SIZE
apachepulsar/pulsar-test-latest-version 2.6.0-SNAPSHOT c599fe91d80e 2 seconds ago 2.82GB
apachepulsar/pulsar-test-latest-version latest c599fe91d80e 2 seconds ago 2.82GB
apachepulsar/pulsar-all latest 77d849fb13b7 3 days ago 2.76GB
node 10 ea119cebc1c3 3 weeks ago 908MB
node 12 6b5991bf650f 3 weeks ago 913MB
buildpack-deps stretch 2adbcf3b6aff 3 weeks ago 835MB
debian 9 f6c68e2ad82a 3 weeks ago 101MB
debian 8 5d7d9c6338e8 3 weeks ago 129MB
node 10-alpine d32bf7fc7d5f 3 weeks ago 80.5MB
node 12-alpine 1cbcaddb8074 3 weeks ago 85.2MB
alpine 3.10 965ea09ff2eb 3 months ago 5.55MB
jekyll/builder latest 4605d94a6b36 3 months ago 564MB
alpine 3.9 055936d39205 8 months ago 5.53MB
alpine 3.7 6d1ef012b567 10 months ago 4.21MB
alpine 3.8 dac705114996 10 months ago 4.41MB
mcr.microsoft.com/azure-pipelines/node8-typescript latest 9a948d360778 15 months ago 595MB
```
### Modifications
* Add groups for integration tests
* Delete no used Docker image
* Pack images as needed
### Verifying this change
Now the integration test can pass normally, please refer to https://github.com/AmateurEvents/pulsar/pull/5/checks?check_run_id=398472038
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 fe03fbb..e112474 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
@@ -81,10 +81,10 @@
@Slf4j
public abstract class PulsarFunctionsTest extends PulsarFunctionsTestBase {
- final Duration ONE_MINUTE = Duration.ofMinutes(1);
- final Duration TEN_SECONDS = Duration.ofSeconds(10);
+ final Duration ONE_MINUTE = Duration.ofMinutes(1);
+ final Duration TEN_SECONDS = Duration.ofSeconds(10);
- final RetryPolicy statusRetryPolicy = new RetryPolicy()
+ final RetryPolicy statusRetryPolicy = new RetryPolicy()
.withMaxDuration(ONE_MINUTE)
.withDelay(TEN_SECONDS)
.onRetry(e -> log.error("Retry ... "));
@@ -93,59 +93,63 @@
super(functionRuntimeType);
}
- @Test
+ @Test(groups = "sink")
public void testKafkaSink() throws Exception {
final String kafkaContainerName = "kafka-" + randomName(8);
testSink(new KafkaSinkTester(kafkaContainerName), true, new KafkaSourceTester(kafkaContainerName));
}
- @Test(enabled = false)
+ @Test(enabled = false, groups = "sink")
public void testCassandraSink() throws Exception {
testSink(CassandraSinkTester.createTester(true), true);
}
- @Test(enabled = false)
+ @Test(enabled = false, groups = "sink")
public void testCassandraArchiveSink() throws Exception {
testSink(CassandraSinkTester.createTester(false), false);
}
- @Test(enabled = false)
+ @Test(enabled = false, groups = "sink")
public void testHdfsSink() throws Exception {
testSink(new HdfsSinkTester(), false);
}
- @Test
+ @Test(groups = "sink")
public void testJdbcSink() throws Exception {
testSink(new JdbcSinkTester(), true);
}
- @Test(enabled = false)
+ @Test(enabled = false, groups = "sink")
public void testElasticSearchSink() throws Exception {
testSink(new ElasticSearchSinkTester(), true);
}
- @Test
+ @Test(groups = "sink")
public void testRabbitMQSink() throws Exception {
final String containerName = "rabbitmq-" + randomName(8);
testSink(new RabbitMQSinkTester(containerName), true, new RabbitMQSourceTester(containerName));
}
- @Test
+ @Test(groups = "source")
public void testDebeziumMySqlSource() throws Exception {
testDebeziumMySqlConnect();
}
- @Test
+ @Test(groups = "source")
public void testDebeziumPostgreSqlSource() throws Exception {
testDebeziumPostgreSqlConnect();
}
- @Test
+ @Test(groups = "source")
public void testDebeziumMongoDbSource() throws Exception{
testDebeziumMongoDbConnect();
}
private void testSink(SinkTester tester, boolean builtin) throws Exception {
+ if (pulsarCluster == null) {
+ super.setupCluster();
+ super.setupFunctionWorkers();
+ }
tester.startServiceContainer(pulsarCluster);
try {
runSinkTester(tester, builtin);
@@ -159,6 +163,10 @@
boolean builtinSink,
SourceTester<ServiceContainerT> sourceTester)
throws Exception {
+ if (pulsarCluster == null) {
+ super.setupCluster();
+ super.setupFunctionWorkers();
+ }
ServiceContainerT serviceContainer = sinkTester.startServiceContainer(pulsarCluster);
try {
runSinkTester(sinkTester, builtinSink);
@@ -830,12 +838,12 @@
}
}
- @Test
+ @Test(groups = "function")
public void testPythonFunctionLocalRun() throws Exception {
testFunctionLocalRun(Runtime.PYTHON);
}
- @Test
+ @Test(groups = "function")
public void testJavaFunctionLocalRun() throws Exception {
testFunctionLocalRun(Runtime.JAVA);
}
@@ -845,6 +853,11 @@
return;
}
+ if (pulsarCluster == null) {
+ super.setupCluster();
+ super.setupFunctionWorkers();
+ }
+
String inputTopicName = "persistent://public/default/test-function-local-run-" + runtime + "-input-" + randomName(8);
String outputTopicName = "test-function-local-run-" + runtime + "-output-" + randomName(8);
@@ -961,6 +974,10 @@
String inputTopicName = "test-" + type + "-count-window-" + functionRuntimeType + "-input-" + randomName(8);
String outputTopicName = "test-" + type + "-count-window-" + functionRuntimeType + "-output-" + randomName(8);
+ if (pulsarCluster == null) {
+ super.setupCluster();
+ super.setupFunctionWorkers();
+ }
try (PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(pulsarCluster.getHttpServiceUrl()).build()) {
admin.topics().createNonPartitionedTopic(inputTopicName);
admin.topics().createNonPartitionedTopic(outputTopicName);
@@ -1053,7 +1070,7 @@
getFunctionInfoNotFound(functionName);
}
- @Test
+ @Test(groups="function")
public void testSlidingCountWindowTest() throws Exception {
String[] EXPECTED_RESULTS = {
"0,1,2,3,4",
@@ -1081,7 +1098,7 @@
testWindowFunction("sliding", EXPECTED_RESULTS);
}
- @Test
+ @Test(groups = "function")
public void testTumblingCountWindowTest() throws Exception {
String[] EXPECTED_RESULTS = {
"0,1,2,3,4,5,6,7,8,9",
@@ -1103,12 +1120,12 @@
// Test CRUD functions on different runtimes.
//
- @Test
+ @Test(groups = "function")
public void testPythonFunctionNegAck() throws Exception {
testFunctionNegAck(Runtime.PYTHON);
}
- @Test
+ @Test(groups = "function")
public void testJavaFunctionNegAck() throws Exception {
testFunctionNegAck(Runtime.JAVA);
}
@@ -1118,6 +1135,11 @@
return;
}
+ if (pulsarCluster == null) {
+ super.setupCluster();
+ super.setupFunctionWorkers();
+ }
+
Schema<?> schema;
if (Runtime.JAVA == runtime) {
schema = Schema.STRING;
@@ -1284,12 +1306,12 @@
checkSubscriptionsCleanup(inputTopicName);
}
- @Test
+ @Test(groups = "function")
public void testPythonPublishFunction() throws Exception {
testPublishFunction(Runtime.PYTHON);
}
- @Test
+ @Test(groups = "function")
public void testJavaPublishFunction() throws Exception {
testPublishFunction(Runtime.JAVA);
}
@@ -1306,6 +1328,11 @@
schema = Schema.BYTES;
}
+ if (pulsarCluster == null) {
+ super.setupCluster();
+ super.setupFunctionWorkers();
+ }
+
String inputTopicName = "persistent://public/default/test-publish-" + runtime + "-input-" + randomName(8);
String outputTopicName = "test-publish-" + runtime + "-output-" + randomName(8);
try (PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(pulsarCluster.getHttpServiceUrl()).build()) {
@@ -1405,7 +1432,7 @@
checkSubscriptionsCleanup(inputTopicName);
}
- @Test
+ @Test(groups = "function")
public void testSerdeFunction() throws Exception {
testCustomSerdeFunction();
}
@@ -1415,6 +1442,11 @@
return;
}
+ if (pulsarCluster == null) {
+ super.setupCluster();
+ super.setupFunctionWorkers();
+ }
+
String inputTopicName = "persistent://public/default/test-serde-java-input-" + randomName(8);
String outputTopicName = "test-publish-serde-output-" + randomName(8);
try (PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(pulsarCluster.getHttpServiceUrl()).build()) {
@@ -1447,32 +1479,32 @@
assertEquals(functionStatus.getInstances().get(0).getStatus().isRunning(), true);
}
- @Test
+ @Test(groups = "function")
public void testPythonExclamationFunction() throws Exception {
testExclamationFunction(Runtime.PYTHON, false, false, false);
}
- @Test
+ @Test(groups = "function")
public void testPythonExclamationFunctionWithExtraDeps() throws Exception {
testExclamationFunction(Runtime.PYTHON, false, false, true);
}
- @Test
+ @Test(groups = "function")
public void testPythonExclamationZipFunction() throws Exception {
testExclamationFunction(Runtime.PYTHON, false, true, false);
}
- @Test
+ @Test(groups = "function")
public void testPythonExclamationTopicPatternFunction() throws Exception {
testExclamationFunction(Runtime.PYTHON, true, false, false);
}
- @Test
+ @Test(groups = "function")
public void testJavaExclamationFunction() throws Exception {
testExclamationFunction(Runtime.JAVA, false, false, false);
}
- @Test
+ @Test(groups = "function")
public void testJavaExclamationTopicPatternFunction() throws Exception {
testExclamationFunction(Runtime.JAVA, true, false, false);
}
@@ -1486,6 +1518,11 @@
return;
}
+ if (pulsarCluster == null) {
+ super.setupCluster();
+ super.setupFunctionWorkers();
+ }
+
Schema<?> schema;
if (Runtime.JAVA == runtime) {
schema = Schema.STRING;
@@ -1839,7 +1876,7 @@
PulsarCluster.ADMIN_SCRIPT,
"topics",
"stats",
- topic);
+ topic);
TopicStats topicStats = new Gson().fromJson(result.getStdout(), TopicStats.class);
assertEquals(topicStats.subscriptions.size(), 0);
@@ -1994,13 +2031,18 @@
assertTrue(result.getStderr().isEmpty());
}
- @Test
+ @Test(groups = "function")
public void testAutoSchemaFunction() throws Exception {
String inputTopicName = "test-autoschema-input-" + randomName(8);
String outputTopicName = "test-autoshcema-output-" + randomName(8);
String functionName = "test-autoschema-fn-" + randomName(8);
final int numMessages = 10;
+ if (pulsarCluster == null) {
+ super.setupCluster();
+ super.setupFunctionWorkers();
+ }
+
// submit the exclamation function
submitFunction(
Runtime.JAVA,
@@ -2072,6 +2114,11 @@
// This is the binlog count that contained in mysql container.
final int numMessages = 47;
+ if (pulsarCluster == null) {
+ super.setupCluster();
+ super.setupFunctionWorkers();
+ }
+
@Cleanup
PulsarClient client = PulsarClient.builder()
.serviceUrl(pulsarCluster.getPlainTextServiceUrl())
@@ -2162,6 +2209,11 @@
// This is the binlog count that contained in postgresql container.
final int numMessages = 26;
+ if (pulsarCluster == null) {
+ super.setupCluster();
+ super.setupFunctionWorkers();
+ }
+
@Cleanup
PulsarClient client = PulsarClient.builder()
.serviceUrl(pulsarCluster.getPlainTextServiceUrl())
@@ -2252,6 +2304,11 @@
// This is the binlog count that contained in mongodb container.
final int numMessages = 17;
+ if (pulsarCluster == null) {
+ super.setupCluster();
+ super.setupFunctionWorkers();
+ }
+
@Cleanup
PulsarClient client = PulsarClient.builder()
.serviceUrl(pulsarCluster.getPlainTextServiceUrl())
