From c72255e457f53b08952c6a6a9b84081996112c2b Mon Sep 17 00:00:00 2001 From: manas-ctds Date: Tue, 28 Apr 2026 16:36:19 +0530 Subject: [PATCH 1/5] Bump transitive dependencies to mitigate vulnerabilities --- pom.xml | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) 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 From 19f683fd9c8a6dac1af67ffeed51b04a45988b9c Mon Sep 17 00:00:00 2001 From: manas-ctds Date: Wed, 29 Apr 2026 17:24:45 +0530 Subject: [PATCH 2/5] Update integration tests to run for pulsar 3.0 and 4.0 --- .github/workflows/github-ci.yml | 44 +++++++++++++++---- .../transforms/tests/DockerTest.java | 25 +++++++---- 2 files changed, 51 insertions(+), 18 deletions(-) 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/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); + } + } } From 9a47f8c0b9e267d43c5fea87f253af33d1103a43 Mon Sep 17 00:00:00 2001 From: priyanshu-ctds Date: Wed, 29 Apr 2026 17:51:47 +0530 Subject: [PATCH 3/5] Added a leader readiness check --- .../transforms/tests/AbstractDockerTest.java | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) 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..8cf9f8c 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 @@ -595,10 +595,22 @@ private void deployFunction( new Gson().fromJson(userConfig, new TypeToken>() {}.getType())) .build(); - admin.functions().createFunction(functionConfig, null); + for (int i = 0; i < 600; i++) { + try { + admin.functions().createFunction(functionConfig, null); + break; + } catch (PulsarAdminException e) { + if (e.getMessage() != null && e.getMessage().contains("Leader not yet ready")) { + log.info("Function worker leader not ready yet, retrying... ({}/600)", i + 1); + Thread.sleep(500); + } else { + throw e; + } + } + } FunctionStatus functionStatus = null; - for (int i = 0; i < 300; i++) { + for (int i = 0; i < 600; i++) { functionStatus = admin.functions().getFunctionStatus("public", "default", functionName); if (functionStatus.getNumRunning() == 1) { break; @@ -613,7 +625,7 @@ private void deployFunction( log.error("Function errored out " + f); } }); - Thread.sleep(100); + Thread.sleep(500); } if (functionStatus.getNumRunning() != 1) { From 4bf45d2e65aa12c11a66d6ab401b303b8fe5e716 Mon Sep 17 00:00:00 2001 From: priyanshu-ctds Date: Thu, 30 Apr 2026 15:25:09 +0530 Subject: [PATCH 4/5] Updated wait strategy and added a scheduler for polling to check function status --- .../transforms/tests/AbstractDockerTest.java | 90 +++++++++++-------- .../transforms/tests/PulsarContainer.java | 4 +- 2 files changed, 56 insertions(+), 38 deletions(-) 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 8cf9f8c..6919a7d 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; @@ -595,42 +596,8 @@ private void deployFunction( new Gson().fromJson(userConfig, new TypeToken>() {}.getType())) .build(); - for (int i = 0; i < 600; i++) { - try { - admin.functions().createFunction(functionConfig, null); - break; - } catch (PulsarAdminException e) { - if (e.getMessage() != null && e.getMessage().contains("Leader not yet ready")) { - log.info("Function worker leader not ready yet, retrying... ({}/600)", i + 1); - Thread.sleep(500); - } else { - throw e; - } - } - } - - FunctionStatus functionStatus = null; - for (int i = 0; i < 600; 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(500); - } - - if (functionStatus.getNumRunning() != 1) { - fail("Function didn't start in time"); - } + admin.functions().createFunction(functionConfig, null); + waitForFunctionRunning(functionName); } @Value @@ -644,4 +611,53 @@ 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/PulsarContainer.java b/tests/src/test/java/com/datastax/oss/pulsar/functions/transforms/tests/PulsarContainer.java index d9adc44..6667413 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,9 @@ 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(); From 26f5a7b721dc96f78cd79f68aaf112b742e1a860 Mon Sep 17 00:00:00 2001 From: priyanshu-ctds Date: Thu, 30 Apr 2026 15:33:11 +0530 Subject: [PATCH 5/5] Fix formatting issue --- .../functions/transforms/tests/AbstractDockerTest.java | 9 +++++++-- .../functions/transforms/tests/PulsarContainer.java | 3 ++- 2 files changed, 9 insertions(+), 3 deletions(-) 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 6919a7d..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 @@ -622,7 +622,9 @@ private void waitForFunctionRunning(String functionName) throws InterruptedExcep FunctionStatus status = admin.functions().getFunctionStatus("public", "default", functionName); - status.getInstances().stream() + status + .getInstances() + .stream() .filter(f -> !StringUtils.isEmpty(f.getStatus().getError())) .findFirst() .ifPresent( @@ -650,7 +652,10 @@ private void waitForFunctionRunning(String functionName) throws InterruptedExcep future.completeExceptionally(e); scheduler.shutdown(); } - }, 0, 2, TimeUnit.SECONDS); + }, + 0, + 2, + TimeUnit.SECONDS); try { future.get(); 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 6667413..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 @@ -55,7 +55,8 @@ public void start() { .withStrategy(Wait.defaultWaitStrategy()) .withStrategy(Wait.forLogMessage(".*Created namespace public/default.*", 1)) .withStrategy(Wait.forLogMessage(".*Function worker service started.*", 1)) - .withStrategy(Wait.forLogMessage(".*FunctionMetaDataManager done becoming leader.*", 1))) + .withStrategy( + Wait.forLogMessage(".*FunctionMetaDataManager done becoming leader.*", 1))) .withLogConsumer( (f) -> { String text = f.getUtf8String().trim();