Merge pull request #535 from mgrenonville/add-batch-receive-policy

Add batchReceivePolicy configuration in consumer config
CleverCloud · Dec 7, 2023 · 49fbd6d · 49fbd6d
2 parents ee13db1 + a913ed0
commit 49fbd6d
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 1 deletion.
diff --git a/pulsar4s-core/src/main/scala/com/sksamuel/pulsar4s/ConsumerConfig.scala b/pulsar4s-core/src/main/scala/com/sksamuel/pulsar4s/ConsumerConfig.scala
@@ -29,5 +29,6 @@ case class ConsumerConfig(subscriptionName: Subscription,
                           enableBatchIndexAcknowledgement: Option[Boolean] = None,
                           additionalProperties: Map[String, AnyRef] = Map.empty,
                           deadLetterPolicy: Option[DeadLetterPolicy] = None,
-                          replicateSubscriptionState: Boolean = false)
+                          replicateSubscriptionState: Boolean = false,
+                          batchReceivePolicy: Option[BatchReceivePolicy] = None)
 
diff --git a/pulsar4s-core/src/main/scala/com/sksamuel/pulsar4s/PulsarClient.scala b/pulsar4s-core/src/main/scala/com/sksamuel/pulsar4s/PulsarClient.scala
@@ -287,6 +287,7 @@ class DefaultPulsarClient(client: org.apache.pulsar.client.api.PulsarClient) ext
     config.ackTimeoutTickTime.foreach { tt => builder.ackTimeoutTickTime(tt._1, tt._2) }
     config.enableBatchIndexAcknowledgement.foreach { enabled => builder.enableBatchIndexAcknowledgment(enabled) }
     config.deadLetterPolicy.foreach(builder.deadLetterPolicy)
+    config.batchReceivePolicy.foreach(builder.batchReceivePolicy)
     config.acknowledgmentGroupTime.foreach { gt => builder.acknowledgmentGroupTime(gt._1, gt._2) }
     if (config.topics.nonEmpty)
       builder.topics(config.topics.map(_.name).asJava)