feat(sources): Pulsar should accept broker config options

The `pulsar-rs` crate has been updated to set broker configuration
options such as maxUnackedMessagesPerConsumer. Pull in the update from
our fork to test it.

Ref: LOG-23305

Author: darinspivey

Cargo.lock

@@ -409,7 +409,7 @@ openssl-probe = { version = "0.1.6", default-features = false }
 ordered-float.workspace = true
 percent-encoding = { version = "2.3.1", default-features = false }
 postgres-openssl = { version = "0.5.1", default-features = false, features = ["runtime"], optional = true }
-pulsar = { git = "https://github.com/streamnative/pulsar-rs", rev = "174070149bab64359f8a31094d7bfb857b045bf1", default-features = false, features = ["tokio-runtime", "auth-oauth2", "flate2", "lz4", "snap", "zstd"], optional = true }
+pulsar = { git = "https://github.com/mezmo/pulsar-rs", rev = "9cae903825bf6203f91198cfa4d42ebc9c099b06", default-features = false, features = ["tokio-runtime", "admin-api", "auth-oauth2", "flate2", "lz4", "snap", "zstd"], optional = true }
 quick-junit = { version = "0.5.1" }
 rand.workspace = true
 rand_distr.workspace = true
@@ -509,7 +509,7 @@ tower-test = "0.4.0"
 vector-lib = { workspace = true, features = ["test"] }
 vrl.workspace = true
 
-temp-env = "0.3.1"
+temp-env = { version = "0.3.6", features = ["async_closure"] }
 wiremock = "0.6.4"
 zstd = { version = "0.13.0", default-features = false }
 

Cargo.toml

@@ -156,6 +156,15 @@ spec:
         namespace: pipeline
         key: pipeline-db-secret-uri-key
 
+  - name: pulsar_admin_url
+    optional: true
+    default: 'http://pulsar-proxy.pulsar.svc.cluster.local'
+    valueFrom:
+      configMapKeyRef:
+        name: publishing-backend
+        namespace: default
+        key: pulsar-http-service-url
+
   strTemplates:
   - | # yaml
     apiVersion: v1
@@ -330,6 +339,9 @@ spec:
                 - name: MEZMO_EVENT_TRACE_ENABLED
                   value: "{{{mezmo_event_trace_enabled}}}"
 
+                - name: PULSAR_ADMIN_URL
+                  value: "{{{pulsar_admin_url}}}"
+
                 - name: DB_CONNECTION_POOL_SIZE
                   value: "{{{db_connection_pool_size}}}"
 

deployment/kubernetes/mtp-vector.yaml.envsubst

@@ -3,10 +3,11 @@ version: '3'
 services:
   pulsar:
     image: docker.io/apachepulsar/pulsar:${CONFIG_VERSION}
-    command: sh -c "bin/apply-config-from-env.py conf/standalone.conf && bin/pulsar standalone"
+    command: sh -c "bin/apply-config-from-env.py --prefix PULSAR_PREFIX_ conf/standalone.conf && bin/pulsar standalone"
     ports:
       - 6650:6650
       - 6651:6651
+      - 8080:8080
     environment:
       - PULSAR_PREFIX_brokerServicePortTls=6651
       - PULSAR_PREFIX_tlsKeyFilePath=/etc/pulsar/certs/pulsar.key.pem