fix(pulsar): Upgrade pulsar 6.5.0 and patch tokio-util

We want to upgrade to pulsar@6.5.0, but it contains an updated
dependency for tokio-util which breaks vector. Vector has maintained
a fork of `tokio` and patched it, so this same patch was applied to the
version that pulsar 6.5.0 uses, which is `tokio-util@0.7.17`.

We are testing this version of pulsar as a potential fix for the
`SlowDown` error which causes partitions to no longer be consumed. We
need consumers to be pulling from all available partitions.

Ref: LOG-22655

Author: darinspivey

Cargo.lock

@@ -386,7 +386,7 @@ openssl-probe = { version = "0.1.6", default-features = false }
 ordered-float = { version = "4.6.0", default-features = false, features = ["serde"] }
 percent-encoding = { version = "2.3.1", default-features = false }
 postgres-openssl = { version = "0.5.1", default-features = false, features = ["runtime"], optional = true }
-pulsar = { version = "6.3.1", default-features = false, features = ["tokio-runtime", "auth-oauth2", "flate2", "lz4", "snap", "zstd"], optional = true }
+pulsar = { version = "6.5.0", default-features = false, features = ["tokio-runtime", "auth-oauth2", "flate2", "lz4", "snap", "zstd"], optional = true }
 quick-junit = { version = "0.5.1" }
 rand.workspace = true
 rand_distr.workspace = true
@@ -492,7 +492,7 @@ zstd = { version = "0.13.0", default-features = false }
 
 [patch.crates-io]
 # The upgrade for `tokio-util` >= 0.6.9 is blocked on https://github.com/vectordotdev/vector/issues/11257.
-tokio-util = { git = "https://github.com/vectordotdev/tokio", branch = "tokio-util-0.7.13-framed-read-continue-on-error" }
+tokio-util = { git = "https://github.com/mezmo/tokio", branch = "tokio-util-0.7.17-framed-read-continue-on-error", version = "0.7.17" }
 nix = { git = "https://github.com/vectordotdev/nix.git", branch = "memfd/gnu/musl" }
 # The `heim` crates depend on `ntapi` 0.3.7 on Windows, but that version has an
 # unaligned access bug fixed in the following revision.

Cargo.toml

@@ -336,6 +336,9 @@ impl PulsarSinkConfig {
             batch_size: self.batch.max_events,
             batch_byte_size: self.batch.max_bytes,
             compression: None,
+            batch_timeout: None,
+            block_queue_if_full: false,
+            routing_policy: None,
         };
 
         match &self.compression {