Fix PulsarIO

The main issue for current PulsarIO.read is

* It is based on Pulsar Reader instead of PulsarConsumer, which then
  do not support acknowledgement

* The while() block in reader DoFn would never return until topic termination,
  this basically means pipeline stuck

* The restriction is on publishTime, and tryClaim assumes its ordering.
  This is not true. reader returning message is ordered on messageId.
  This is a wrong choice. Currently unresolved

* PulsarMessage's coder implementation dropped message. This causes
  Data loss if the PulsarIO.read do not follow an immediate mapping

* Tests are defunct and errors are suppressed, making them succeed spuriously

Current PulsarIO.write is even more primitive. Pipeline expansion actually fails. It is not idempotent.

Major fixes include

* Allow Pulsar reader to have a timeout

* Fix PulsarMessage and coder to include serializable fields from message

* Fix mock client/reader and add a full read pipeline in test

* Fix issues prevent PulsarIO.write from expanding. now it works minimally,
  that is publish every message received (at least once).

* Working integration tests for read and write

This has made PulsarIO.read minimally functionable. Although it won't split
and can only run single thread.

Going forward, we should re-implement reader DoFn based on Pulsar consumer.
Thoughs rename the current DoFn to "NaiveReadFromPulsarDoFn"

Author: Abacn

.github/workflows/beam_PreCommit_Java_Pulsar_IO_Direct.yml

@@ -21,31 +21,13 @@ on:
     branches: ['master', 'release-*']
     paths:
       - "sdks/java/io/pulsar/**"
-      - "sdks/java/io/common/**"
-      - "sdks/java/core/src/main/**"
-      - "build.gradle"
-      - "buildSrc/**"
-      - "gradle/**"
-      - "gradle.properties"
-      - "gradlew"
-      - "gradle.bat"
-      - "settings.gradle.kts"
       - ".github/workflows/beam_PreCommit_Java_Pulsar_IO_Direct.yml"
   pull_request_target:
     branches: ['master', 'release-*']
     paths:
       - "sdks/java/io/pulsar/**"
-      - "sdks/java/io/common/**"
-      - "sdks/java/core/src/main/**"
+      - ".github/workflows/beam_PreCommit_Java_Pulsar_IO_Direct.yml"
       - 'release/trigger_all_tests.json'
-      - '.github/trigger_files/beam_PreCommit_Java_Pulsar_IO_Direct.json'
-      - "build.gradle"
-      - "buildSrc/**"
-      - "gradle/**"
-      - "gradle.properties"
-      - "gradlew"
-      - "gradle.bat"
-      - "settings.gradle.kts"
   issue_comment:
     types: [created]
   schedule:
@@ -110,6 +92,13 @@ jobs:
           arguments: |
             -PdisableSpotlessCheck=true \
             -PdisableCheckStyle=true \
+      - name: run Pulsar IO IT script
+        uses: ./.github/actions/gradle-command-self-hosted-action
+        with:
+          gradle-command: :sdks:java:io:pulsar:integrationTest
+          arguments: |
+            -PdisableSpotlessCheck=true \
+            -PdisableCheckStyle=true \
       - name: Archive JUnit Test Results
         uses: actions/upload-artifact@v4
         if: ${{ !success() }}
@@ -135,4 +124,4 @@ jobs:
         if: always()
         with:
           name: Publish SpotBugs
-          path: '**/build/reports/spotbugs/*.html'
+          path: '**/build/reports/spotbugs/*.html'

build.gradle.kts

@@ -355,6 +355,7 @@ tasks.register("javaioPreCommit") {
   dependsOn(":sdks:java:io:mqtt:build")
   dependsOn(":sdks:java:io:neo4j:build")
   dependsOn(":sdks:java:io:parquet:build")
+  dependsOn(":sdks:java:io:pulsar:build")
   dependsOn(":sdks:java:io:rabbitmq:build")
   dependsOn(":sdks:java:io:redis:build")
   dependsOn(":sdks:java:io:rrio:build")

sdks/java/io/pulsar/build.gradle

@@ -18,31 +18,32 @@
 
 plugins { id 'org.apache.beam.module' }
 applyJavaNature(automaticModuleName: 'org.apache.beam.sdk.io.pulsar')
+enableJavaPerformanceTesting()
 
 description = "Apache Beam :: SDKs :: Java :: IO :: Pulsar"
 ext.summary = "IO to read and write to Pulsar"
 
-def pulsar_version = '2.8.2'
+def pulsar_version = '2.11.4'
 
 
 dependencies {
     implementation library.java.vendored_guava_32_1_2_jre
     implementation library.java.slf4j_api
     implementation library.java.joda_time
 
-    implementation "org.apache.pulsar:pulsar-client:$pulsar_version"
-    implementation "org.apache.pulsar:pulsar-client-admin:$pulsar_version"
-    permitUnusedDeclared "org.apache.pulsar:pulsar-client:$pulsar_version"
-    permitUnusedDeclared "org.apache.pulsar:pulsar-client-admin:$pulsar_version"
-    permitUsedUndeclared "org.apache.pulsar:pulsar-client-api:$pulsar_version"
-    permitUsedUndeclared "org.apache.pulsar:pulsar-client-admin-api:$pulsar_version"
+    implementation "org.apache.pulsar:pulsar-client-api:$pulsar_version"
+    implementation "org.apache.pulsar:pulsar-client-admin-api:$pulsar_version"
+    runtimeOnly "org.apache.pulsar:pulsar-client:$pulsar_version"
+    runtimeOnly("org.apache.pulsar:pulsar-client-admin:$pulsar_version") {
+      // To prevent a StackOverflow within Pulsar admin client because JUL -> SLF4J -> JUL
+      exclude group: "org.slf4j", module: "jul-to-slf4j"
+    }
 
     implementation project(path: ":sdks:java:core", configuration: "shadow")
 
-    testImplementation library.java.jupiter_api
-    testRuntimeOnly library.java.jupiter_engine
+    testImplementation library.java.junit
+    testRuntimeOnly library.java.slf4j_jdk14
     testRuntimeOnly project(path: ":runners:direct-java", configuration: "shadow")
     testImplementation "org.testcontainers:pulsar:1.15.3"
     testImplementation "org.assertj:assertj-core:2.9.1"
-
 }