diff --git a/.github/workflows/github-ci.yml b/.github/workflows/github-ci.yml index c004e35..27bb90b 100644 --- a/.github/workflows/github-ci.yml +++ b/.github/workflows/github-ci.yml @@ -35,8 +35,8 @@ jobs: - name: Unit tests run: mvn -B test -DfailIfNoTests=false -pl streaming-ai,pulsar-transformations,pulsar-ai-tools -Dspotbugs.skip -Dlicense.skip -Dfmt.skip -Dxml-format.skip - tests-ls: - name: Integration tests on Luna Streaming + tests-ls-31: + name: Integration tests on Luna Streaming 3.1 runs-on: ubuntu-latest needs: build steps: @@ -58,11 +58,11 @@ jobs: - name: Build NAR run: mvn -B clean package -pl streaming-ai,pulsar-transformations -DskipTests -Dspotbugs.skip -Dlicense.skip -Dfmt.skip -Dxml-format.skip - - name: Integration tests on Luna Streaming 2.10 - run: mvn -B test -DfailIfNoTests=false -pl tests -Dtest=DockerTest\$LunaStreaming210Test -Dspotbugs.skip -Dlicense.skip -Dfmt.skip -Dxml-format.skip + - name: Integration tests on Luna Streaming 3.1 + run: mvn -B test -DfailIfNoTests=false -pl tests -Dtest=DockerTest\$LunaStreaming31Test -Dspotbugs.skip -Dlicense.skip -Dfmt.skip -Dxml-format.skip - tests-pulsar-211: - name: Integration tests on Pulsar 2.11 + tests-ls-40: + name: Integration tests on Luna Streaming 4.0 runs-on: ubuntu-latest needs: build steps: @@ -84,10 +84,10 @@ jobs: - name: Build NAR run: mvn -B clean package -pl streaming-ai,pulsar-transformations -DskipTests -Dspotbugs.skip -Dlicense.skip -Dfmt.skip -Dxml-format.skip - - name: Integration tests on Pulsar 2.11 - run: mvn -B test -DfailIfNoTests=false -pl tests -Dtest=DockerTest\$Pulsar211Test -Dspotbugs.skip -Dlicense.skip -Dfmt.skip -Dxml-format.skip + - name: Integration tests on Luna Streaming 4.0 + run: mvn -B test -DfailIfNoTests=false -pl tests -Dtest=DockerTest\$LunaStreaming40Test -Dspotbugs.skip -Dlicense.skip -Dfmt.skip -Dxml-format.skip - tests-pulsar-3: + tests-pulsar-30: name: Integration tests on Pulsar 3.0 runs-on: ubuntu-latest needs: build @@ -112,3 +112,29 @@ jobs: - name: Integration tests on Pulsar 3.0 run: mvn -B test -DfailIfNoTests=false -pl tests -Dtest=DockerTest\$Pulsar30Test -Dspotbugs.skip -Dlicense.skip -Dfmt.skip -Dxml-format.skip + + tests-pulsar-40: + name: Integration tests on Pulsar 4.0 + runs-on: ubuntu-latest + needs: build + steps: + - uses: actions/checkout@v2 + - name: Set up JDK 11 + uses: actions/setup-java@v2 + with: + java-version: '11' + distribution: 'adopt' + + - name: Cache local Maven repository + uses: actions/cache@v3 + with: + path: ~/.m2/repository + key: ${{ runner.os }}-maven-${{ hashFiles('**/pom.xml') }} + restore-keys: | + ${{ runner.os }}-maven- + + - name: Build NAR + run: mvn -B clean package -pl streaming-ai,pulsar-transformations -DskipTests -Dspotbugs.skip -Dlicense.skip -Dfmt.skip -Dxml-format.skip + + - name: Integration tests on Pulsar 4.0 + run: mvn -B test -DfailIfNoTests=false -pl tests -Dtest=DockerTest\$Pulsar40Test -Dspotbugs.skip -Dlicense.skip -Dfmt.skip -Dxml-format.skip diff --git a/pom.xml b/pom.xml index 672e7de..12ed5c9 100644 --- a/pom.xml +++ b/pom.xml @@ -53,11 +53,11 @@ 3.18.0 1.26.0 2.14.0 - 2.12.4 - 0.27 - 2.15.4 + 2.14.5 + 2.0.3 + 2.21.2 2.0 - 4.1.129.Final + 4.1.132.Final streaming-ai diff --git a/tests/src/test/java/com/datastax/oss/pulsar/functions/transforms/tests/AbstractDockerTest.java b/tests/src/test/java/com/datastax/oss/pulsar/functions/transforms/tests/AbstractDockerTest.java index 8fc4ff9..bdfbf1e 100644 --- a/tests/src/test/java/com/datastax/oss/pulsar/functions/transforms/tests/AbstractDockerTest.java +++ b/tests/src/test/java/com/datastax/oss/pulsar/functions/transforms/tests/AbstractDockerTest.java @@ -25,12 +25,13 @@ import com.google.gson.Gson; import com.google.gson.reflect.TypeToken; import java.math.BigDecimal; +import java.time.Duration; import java.time.LocalDate; import java.util.Collections; import java.util.List; import java.util.Map; import java.util.UUID; -import java.util.concurrent.TimeUnit; +import java.util.concurrent.*; import lombok.Value; import lombok.extern.slf4j.Slf4j; import org.apache.avro.Conversions; @@ -596,29 +597,7 @@ private void deployFunction( .build(); admin.functions().createFunction(functionConfig, null); - - FunctionStatus functionStatus = null; - for (int i = 0; i < 300; i++) { - functionStatus = admin.functions().getFunctionStatus("public", "default", functionName); - if (functionStatus.getNumRunning() == 1) { - break; - } - log.info("Function status: {}", functionStatus); - functionStatus - .getInstances() - .forEach( - f -> { - log.info("Function instance status: {}", f); - if (!StringUtils.isEmpty(f.getStatus().getError())) { - log.error("Function errored out " + f); - } - }); - Thread.sleep(100); - } - - if (functionStatus.getNumRunning() != 1) { - fail("Function didn't start in time"); - } + waitForFunctionRunning(functionName); } @Value @@ -632,4 +611,58 @@ private static class Pojo2 { String c; String d; } + + private void waitForFunctionRunning(String functionName) throws InterruptedException { + ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(); + CompletableFuture future = new CompletableFuture<>(); + long deadline = System.currentTimeMillis() + Duration.ofMinutes(5).toMillis(); + scheduler.scheduleWithFixedDelay( + () -> { + try { + FunctionStatus status = + admin.functions().getFunctionStatus("public", "default", functionName); + + status + .getInstances() + .stream() + .filter(f -> !StringUtils.isEmpty(f.getStatus().getError())) + .findFirst() + .ifPresent( + f -> + future.completeExceptionally( + new RuntimeException("Function errored out: " + f))); + + if (future.isCompletedExceptionally()) { + scheduler.shutdown(); + return; + } + if (status.getNumRunning() == 1) { + future.complete(null); + scheduler.shutdown(); + return; + } + if (System.currentTimeMillis() > deadline) { + future.completeExceptionally( + new RuntimeException("Function didn't start in time. Status: " + status)); + scheduler.shutdown(); + return; + } + log.info("Function not running yet, status: {}", status); + } catch (PulsarAdminException e) { + future.completeExceptionally(e); + scheduler.shutdown(); + } + }, + 0, + 2, + TimeUnit.SECONDS); + + try { + future.get(); + } catch (ExecutionException e) { + fail(e.getCause().getMessage()); + } finally { + scheduler.shutdownNow(); + } + } } diff --git a/tests/src/test/java/com/datastax/oss/pulsar/functions/transforms/tests/DockerTest.java b/tests/src/test/java/com/datastax/oss/pulsar/functions/transforms/tests/DockerTest.java index 959a223..341f664 100644 --- a/tests/src/test/java/com/datastax/oss/pulsar/functions/transforms/tests/DockerTest.java +++ b/tests/src/test/java/com/datastax/oss/pulsar/functions/transforms/tests/DockerTest.java @@ -17,19 +17,20 @@ public class DockerTest { - private static final String IMAGE_LUNASTREAMING210 = "datastax/lunastreaming:2.10_4.7"; - private static final String IMAGE_PULSAR211 = "apachepulsar/pulsar:2.11.1"; - private static final String IMAGE_PULSAR30 = "apachepulsar/pulsar:3.0.0"; + private static final String IMAGE_LUNASTREAMING31 = "datastax/lunastreaming:3.1.4_26"; + private static final String IMAGE_LUNASTREAMING40 = "datastax/lunastreaming:4.0.7_7"; + private static final String IMAGE_PULSAR30 = "apachepulsar/pulsar:3.0.17"; + private static final String IMAGE_PULSAR40 = "apachepulsar/pulsar:4.0.10"; - public static class LunaStreaming210Test extends AbstractDockerTest { - LunaStreaming210Test() { - super(IMAGE_LUNASTREAMING210); + public static class LunaStreaming31Test extends AbstractDockerTest { + LunaStreaming31Test() { + super(IMAGE_LUNASTREAMING31); } } - public static class Pulsar211Test extends AbstractDockerTest { - Pulsar211Test() { - super(IMAGE_PULSAR211); + public static class LunaStreaming40Test extends AbstractDockerTest { + LunaStreaming40Test() { + super(IMAGE_LUNASTREAMING40); } } @@ -38,4 +39,10 @@ public static class Pulsar30Test extends AbstractDockerTest { super(IMAGE_PULSAR30); } } + + public static class Pulsar40Test extends AbstractDockerTest { + Pulsar40Test() { + super(IMAGE_PULSAR40); + } + } } diff --git a/tests/src/test/java/com/datastax/oss/pulsar/functions/transforms/tests/PulsarContainer.java b/tests/src/test/java/com/datastax/oss/pulsar/functions/transforms/tests/PulsarContainer.java index d9adc44..d72aca0 100644 --- a/tests/src/test/java/com/datastax/oss/pulsar/functions/transforms/tests/PulsarContainer.java +++ b/tests/src/test/java/com/datastax/oss/pulsar/functions/transforms/tests/PulsarContainer.java @@ -53,7 +53,10 @@ public void start() { .waitingFor( (new WaitAllStrategy()) .withStrategy(Wait.defaultWaitStrategy()) - .withStrategy(Wait.forLogMessage(".*Created namespace public/default.*", 1))) + .withStrategy(Wait.forLogMessage(".*Created namespace public/default.*", 1)) + .withStrategy(Wait.forLogMessage(".*Function worker service started.*", 1)) + .withStrategy( + Wait.forLogMessage(".*FunctionMetaDataManager done becoming leader.*", 1))) .withLogConsumer( (f) -> { String text = f.getUtf8String().trim();