Update pulsar dependency to 2.10.3

agent/src/main/java/com/datastax/oss/cdc/agent/AbstractPulsarMutationSender.java

@@ -92,6 +92,7 @@ public void initialize(AgentConfig config) throws PulsarClientException {
         try {
             ClientBuilder clientBuilder = PulsarClient.builder()
                     .serviceUrl(config.pulsarServiceUrl)
+                    .memoryLimit(config.pulsarMemoryLimitBytes, SizeUnit.BYTES)
                     .enableTcpNoDelay(false);
 
             if (config.pulsarServiceUrl.startsWith("pulsar+ssl://")) {
@@ -197,7 +198,6 @@ public Producer<KeyValue<byte[], MutationValue>> getProducer(final TableInfo tm)
                         .hashingScheme(HashingScheme.Murmur3_32Hash)
                         .blockIfQueueFull(true)
                         .maxPendingMessages(config.pulsarMaxPendingMessages)
-                        .maxPendingMessagesAcrossPartitions(config.pulsarMaxPendingMessagesAcrossPartitions)
                         .autoUpdatePartitions(true);
 
                 if (config.pulsarBatchDelayInMs > 0) {

agent/src/main/java/com/datastax/oss/cdc/agent/AgentConfig.java

@@ -297,13 +297,13 @@ public static long getEnvAsLong(String varName, long defaultValue) {
                     1000, "CDC_PULSAR_MAX_PENDING_MESSAGES", Setting::getEnvAsInteger,
                     "Integer", "pulsar", 4);
 
-    public static final String PULSAR_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS= "pulsarMaxPendingMessagesAcrossPartitions";
-    public int pulsarMaxPendingMessagesAcrossPartitions;
-    public static final Setting<Integer> PULSAR_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS_SETTING =
-            new Setting<>(PULSAR_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS, Platform.PULSAR, (c, s) -> c.pulsarMaxPendingMessagesAcrossPartitions = Integer.parseInt(s), c -> c.pulsarMaxPendingMessagesAcrossPartitions,
-                    "The Pulsar maximum number of pending messages across partitions.",
-                    50000, "CDC_PULSAR_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS", Setting::getEnvAsInteger,
-                    "Integer", "pulsar", 5);
+    public static final String PULSAR_MEMORY_LIMIT_BYTES= "pulsarMemoryLimitBytes";
+    public long pulsarMemoryLimitBytes;
+    public static final Setting<Long> PULSAR_MEMORY_LIMIT_BYTES_SETTING =
+            new Setting<>(PULSAR_MEMORY_LIMIT_BYTES, Platform.PULSAR, (c, s) -> c.pulsarMemoryLimitBytes = Long.parseLong(s), c -> c.pulsarMemoryLimitBytes,
+                    "Limit of client memory usage (in bytes). The 0 default means memory limit is disabled.",
+                    0L, "CDC_PULSAR_MEMORY_LIMIT_BYTES", Setting::getEnvAsLong,
+                    "Long", "pulsar", 5);
 
     public static final String PULSAR_AUTH_PLUGIN_CLASS_NAME = "pulsarAuthPluginClassName";
     public String pulsarAuthPluginClassName;
@@ -349,9 +349,9 @@ public static long getEnvAsLong(String varName, long defaultValue) {
         set.add(PULSAR_BATCH_BATCH_DELAY_IN_MS_SETTING);
         set.add(PULSAR_KEY_BASED_BATCHER_SETTING);
         set.add(PULSAR_MAX_PENDING_MESSAGES_SETTING);
-        set.add(PULSAR_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS_SETTING);
         set.add(PULSAR_AUTH_PLUGIN_CLASS_NAME_SETTING);
         set.add(PULSAR_AUTH_PARAMS_SETTING);
+        set.add(PULSAR_MEMORY_LIMIT_BYTES_SETTING);
         settings = Collections.unmodifiableSet(set);
 
         Map<String, Setting<?>> map = new HashMap<>();
@@ -382,9 +382,9 @@ public AgentConfig() {
         this.pulsarBatchDelayInMs = PULSAR_BATCH_BATCH_DELAY_IN_MS_SETTING.initDefault();
         this.pulsarKeyBasedBatcher = PULSAR_KEY_BASED_BATCHER_SETTING.initDefault();
         this.pulsarMaxPendingMessages = PULSAR_MAX_PENDING_MESSAGES_SETTING.initDefault();
-        this.pulsarMaxPendingMessagesAcrossPartitions = PULSAR_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS_SETTING.initDefault();
         this.pulsarAuthPluginClassName = PULSAR_AUTH_PLUGIN_CLASS_NAME_SETTING.initDefault();
         this.pulsarAuthParams = PULSAR_AUTH_PARAMS_SETTING.initDefault();
+        this.pulsarMemoryLimitBytes = PULSAR_MEMORY_LIMIT_BYTES_SETTING.initDefault();
     }
 
     public static void main(String[] args) {
@@ -509,7 +509,10 @@ public AgentConfig configure(Platform platform, Map<String, Object> agentParamet
                     throw new IllegalArgumentException(String.format("Unsupported parameter '%s' for the %s platform ", key, platform));
                 }
                 setting.initializer.apply(this, value);
-            } else {
+            } else if ("pulsarMaxPendingMessagesAcrossPartitions".equals(key)) {
+                log.warn("The 'pulsarMaxPendingMessagesAcrossPartitions' parameter is deprecated, the config will be ignored");
+            }
+            else {
                 throw new RuntimeException(String.format("Unknown parameter '%s'", key));
             }
         }

agent/src/test/java/com/datastax/oss/cdc/agent/AgentParametersTest.java

@@ -71,10 +71,11 @@ void assertCommonConfig(AgentConfig config) {
     public void testConfigurePulsar() {
         String agentArgs = COMMON_CONFIG +
                 PULSAR_SERVICE_URL + "=pulsar+ssl://mypulsar:6650\\,localhost:6651\\,localhost:6652," +
+                PULSAR_MEMORY_LIMIT_BYTES + "=64," +
                 PULSAR_BATCH_DELAY_IN_MS + "=20," +
                 PULSAR_KEY_BASED_BATCHER + "=true," +
                 PULSAR_MAX_PENDING_MESSAGES + "=20," +
-                PULSAR_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS + "=200," +
+                "pulsarMaxPendingMessagesAcrossPartitions" + "=200," + // make sure if someone is passing the deprecated parameter, AgentConfig will not fail
                 PULSAR_AUTH_PLUGIN_CLASS_NAME + "=MyAuthPlugin," +
                 PULSAR_AUTH_PARAMS + "=x:y\\,z:t," +
                 SSL_ALLOW_INSECURE_CONNECTION + "=true," +
@@ -91,7 +92,7 @@ public void testConfigurePulsar() {
         assertEquals(20L, config.pulsarBatchDelayInMs);
         assertTrue(config.pulsarKeyBasedBatcher);
         assertEquals(20, config.pulsarMaxPendingMessages);
-        assertEquals(200, config.pulsarMaxPendingMessagesAcrossPartitions);
+        assertEquals(64L, config.pulsarMemoryLimitBytes);
 
         // Pulsar Auth
         assertEquals("MyAuthPlugin", config.pulsarAuthPluginClassName);
@@ -107,9 +108,11 @@ public void testConfigurePulsarFromMap() {
         tenantInfo.put(SSL_ALLOW_INSECURE_CONNECTION, "true");
         tenantInfo.put(SSL_HOSTNAME_VERIFICATION_ENABLE, "true");
         tenantInfo.put(TLS_TRUST_CERTS_FILE_PATH, "/test.p12");
+        tenantInfo.put(PULSAR_MEMORY_LIMIT_BYTES, "64");
 
         AgentConfig config = AgentConfig.create(Platform.PULSAR, tenantInfo);
         assertEquals("pulsar+ssl://mypulsar:6650,localhost:6651,localhost:6652", config.pulsarServiceUrl);
+        assertEquals(64L, config.pulsarMemoryLimitBytes);
 
         // Pulsar Auth
         assertEquals("MyAuthPlugin", config.pulsarAuthPluginClassName);
@@ -129,6 +132,7 @@ public void testConfigurePulsarFromMap() {
     @SetEnvironmentVariable(key = "CDC_PULSAR_BATCH_DELAY_IN_MS", value = "555")
     @SetEnvironmentVariable(key = "CDC_PULSAR_KEY_BASED_BATCHER", value = "true")
     @SetEnvironmentVariable(key = "CDC_PULSAR_MAX_PENDING_MESSAGES", value = "555")
+    @SetEnvironmentVariable(key = "CDC_PULSAR_MEMORY_LIMIT_BYTES", value = "64")
     public void testConfigurePulsarFromEnvVar() {
         AgentConfig config = AgentConfig.create(Platform.PULSAR, "");
         assertEquals("pulsar+ssl://mypulsar:6650,localhost:6651,localhost:6652", config.pulsarServiceUrl);
@@ -139,6 +143,7 @@ public void testConfigurePulsarFromEnvVar() {
         assertEquals(555L, config.pulsarBatchDelayInMs);
         assertEquals(true, config.pulsarKeyBasedBatcher);
         assertEquals(555, config.pulsarMaxPendingMessages);
+        assertEquals(64L, config.pulsarMemoryLimitBytes);
 
         // Pulsar Auth
         assertEquals("MyAuthPlugin", config.pulsarAuthPluginClassName);

connector/build.gradle

@@ -71,7 +71,11 @@ dependencies {
 
     testImplementation "org.testcontainers:testcontainers:${testContainersVersion}"
     testImplementation project(':testcontainers')
-    testImplementation("${pulsarGroup}:pulsar-client:${pulsarVersion}")
+    testImplementation("${pulsarGroup}:pulsar-client") {
+        version {
+            strictly("${pulsarVersion}")
+        }
+    }
 
     nar "${pulsarGroup}:pulsar-io:${pulsarVersion}"
 }

connector/gradle.properties

@@ -1 +1,5 @@
 artifact=connector
+
+# TODO: remove this to fall back to the oveerall client used in the project (2.20.3) once the following issue is fixed:
+# https://github.com/apache/pulsar/issues/20092
+pulsarVersion=2.8.3

gradle.properties

@@ -15,9 +15,10 @@ cassandra4Version=4.0.4
 dse4Version=6.8.23
 
 pulsarGroup=org.apache.pulsar
-pulsarVersion=2.8.3
+pulsarVersion=2.10.3
+# Used when running tests locally, CI will override those values
 testPulsarImage=datastax/lunastreaming
-testPulsarImageTag=2.8.0_1.1.42
+testPulsarImageTag=2.10_3.4
 
 kafkaVersion=3.4.0
 vavrVersion=0.10.3