Fix PulsarIO (#36141)

* 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"

* Update CHANGES.md

* Fix CHANGES.md lint support multi-line item

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'

CHANGES.md

@@ -88,6 +88,9 @@
 ## Bugfixes
 
 * Fixed X (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
+* PulsarIO has now changed support status from incomplete to experimental. Both read and writes should now minimally
+  function (un-partitioned topics, without schema support, timestamp ordered messages for read) (Java)
+  ([#36141](https://github.com/apache/beam/issues/36141)).
 
 ## Known Issues
 
@@ -133,7 +136,7 @@
   significant digits related to casting.
 * (Python) The deterministic fallback coder for complex types like NamedTuple, Enum, and dataclasses now uses cloudpickle instead of dill. If your pipeline is affected, you may see a warning like: "Using fallback deterministic coder for type X...". You can revert to the previous behavior by using the pipeline option `--update_compatibility_version=2.67.0` ([35725](https://github.com/apache/beam/pull/35725)). Report any pickling related issues to [#34903](https://github.com/apache/beam/issues/34903)
 * (Python) Prism runner now enabled by default for most Python pipelines using the direct runner ([#34612](https://github.com/apache/beam/pull/34612)). This may break some tests, see https://github.com/apache/beam/pull/34612 for details on how to handle issues.
-* Dropped Java 8 support for [IO expansion-service](https://central.sonatype.com/artifact/org.apache.beam/beam-sdks-java-io-expansion-service). Cross-language pipelines using this expansion service will need a Java11+ runtime ([#35981](https://github.com/apache/beam/pull/35981).
+* Dropped Java 8 support for [IO expansion-service](https://central.sonatype.com/artifact/org.apache.beam/beam-sdks-java-io-expansion-service). Cross-language pipelines using this expansion service will need a Java11+ runtime ([#35981](https://github.com/apache/beam/pull/35981)).
 
 ## Deprecations
 

build.gradle.kts

@@ -1,3 +1,5 @@
+import java.util.TreeMap
+
 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
@@ -355,6 +357,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")
@@ -691,12 +694,31 @@ tasks.register("validateChanges") {
 
     // Check entries in the unreleased section
     var i = unreleasedSectionStart + 1
-    println("Starting validation from line ${i+1}")
-
+    val items = TreeMap<Int, String>()
+    var lastline = 0
+    var item = ""
     while (i < lines.size && !lines[i].startsWith("# [")) {
       val line = lines[i].trim()
+      if (line.isEmpty()) {
+        // skip
+      } else if (line.startsWith("* ")) {
+        items.put(lastline, item)
+        lastline = i
+        item = line
+      } else if (line.startsWith("##")) {
+        items.put(lastline, item)
+        lastline = i
+        item = ""
+      } else {
+        item += line
+      }
+      i++
+    }
+    items.put(lastline, item)
+    println("Starting validation from line ${i+1}")
 
-      if (line.startsWith("* ") && line.isNotEmpty()) {
+    items.forEach { (i, line) ->
+      if (line.startsWith("* ")) {
         println("Checking line ${i+1}: $line")
 
         // Skip comment lines
@@ -747,8 +769,6 @@ tasks.register("validateChanges") {
           }
         }
       }
-
-      i++
     }
 
     println("Found ${errors.size} errors")

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"
-
 }