Make it possible to enable functions worker in PulsarContainer (#2711)

* Method to enable functions worker in pulsar container

This change should simplify a bit pulsar functions development experience
Tried here: https://github.com/lanwen/pulsar-functions-example

Without waiter pulsar could return random 503, depending on how fast
was a call.

* treat functions worker method as a flag to enable in configure

real functions upload test

* upd pulsar to 2.5.1

* don't store jar

* expect error for functions without worker

* reduce the changeset

Co-authored-by: Sergei Egorov <[email protected]>

Author: lanwen

modules/pulsar/build.gradle

@@ -5,4 +5,5 @@ dependencies {
 
     testCompile group: 'org.apache.pulsar', name: 'pulsar-client', version: '2.6.0'
     testCompile group: 'org.assertj', name: 'assertj-core', version: '3.16.1'
+    testCompile group: 'org.apache.pulsar', name: 'pulsar-client-admin', version: '2.5.1'
 }

modules/pulsar/src/main/java/org/testcontainers/containers/PulsarContainer.java

@@ -1,6 +1,7 @@
 package org.testcontainers.containers;
 
 import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.containers.wait.strategy.WaitAllStrategy;
 import org.testcontainers.utility.TestcontainersConfiguration;
 
 /**
@@ -14,6 +15,8 @@ public class PulsarContainer extends GenericContainer<PulsarContainer> {
 
     private static final String PULSAR_VERSION = "2.2.0";
 
+    private boolean functionsWorkerEnabled = false;
+
     public PulsarContainer() {
         this(PULSAR_VERSION);
     }
@@ -25,6 +28,25 @@ public PulsarContainer(String pulsarVersion) {
         waitingFor(Wait.forHttp(METRICS_ENDPOINT).forStatusCode(200).forPort(BROKER_HTTP_PORT));
     }
 
+    @Override
+    protected void configure() {
+        super.configure();
+
+        if (functionsWorkerEnabled) {
+            withCommand("/pulsar/bin/pulsar", "standalone");
+            waitingFor(
+                new WaitAllStrategy()
+                    .withStrategy(waitStrategy)
+                    .withStrategy(Wait.forLogMessage(".*Function worker service started.*", 1))
+            );
+        }
+    }
+
+    public PulsarContainer withFunctionsWorker() {
+        functionsWorkerEnabled = true;
+        return this;
+    }
+
     public String getPulsarBrokerUrl() {
         return String.format("pulsar://%s:%s", getHost(), getMappedPort(BROKER_PORT));
     }
@@ -33,4 +55,3 @@ public String getHttpServiceUrl() {
         return String.format("http://%s:%s", getHost(), getMappedPort(BROKER_HTTP_PORT));
     }
 }
-

modules/pulsar/src/test/java/org/testcontainers/containers/PulsarContainerTest.java

@@ -1,5 +1,7 @@
 package org.testcontainers.containers;
 
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.Producer;
@@ -10,6 +12,7 @@
 import java.util.concurrent.TimeUnit;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 public class PulsarContainerTest {
 
@@ -23,6 +26,33 @@ public void testUsage() throws Exception {
         }
     }
 
+    @Test
+    public void shouldNotEnableFunctionsWorkerByDefault() throws Exception {
+        try (PulsarContainer pulsar = new PulsarContainer("2.5.1")) {
+            pulsar.start();
+
+            PulsarAdmin pulsarAdmin = PulsarAdmin.builder()
+                .serviceHttpUrl(pulsar.getHttpServiceUrl())
+                .build();
+
+            assertThatThrownBy(() -> pulsarAdmin.functions().getFunctions("public", "default"))
+                .isInstanceOf(PulsarAdminException.class);
+        }
+    }
+
+    @Test
+    public void shouldWaitForFunctionsWorkerStarted() throws Exception {
+        try (PulsarContainer pulsar = new PulsarContainer("2.5.1").withFunctionsWorker()) {
+            pulsar.start();
+
+            PulsarAdmin pulsarAdmin = PulsarAdmin.builder()
+                .serviceHttpUrl(pulsar.getHttpServiceUrl())
+                .build();
+
+            assertThat(pulsarAdmin.functions().getFunctions("public", "default")).hasSize(0);
+        }
+    }
+
     protected void testPulsarFunctionality(String pulsarBrokerUrl) throws Exception {
 
         try (