Add Pulsar reconsumeLater API

pulsar4s-cats-effect/src/main/scala/com/sksamuel/pulsar4s/cats/CatsAsyncHandler.scala

@@ -197,6 +197,11 @@ trait CatsAsyncHandlerLowPriority {
 
     def abortTransaction(txn: Transaction): F[Unit] = Async[F].delay(txn.abort()).liftF.map(_ => ())
 
+    override def reconsumeLaterAsync[T](consumer: JConsumer[T], message: ConsumerMessage[T], delayTime: Long, unit: TimeUnit)
+                                       (implicit schema: Schema[T]): F[Unit] =
+      Async[F].delay {
+        consumer.reconsumeLater(ConsumerMessage.toJava(message, schema), delayTime, unit)
+      }
   }
 
 }

pulsar4s-core/src/main/scala/com/sksamuel/pulsar4s/AsyncHandler.scala

@@ -1,9 +1,10 @@
 package com.sksamuel.pulsar4s
 
 import org.apache.pulsar.client.api
-import org.apache.pulsar.client.api.TypedMessageBuilder
+import org.apache.pulsar.client.api.{Schema, TypedMessageBuilder}
 import org.apache.pulsar.client.api.transaction.Transaction
 
+import java.util.concurrent.TimeUnit
 import scala.concurrent.{ExecutionContext, Future}
 import scala.util.Try
 
@@ -47,6 +48,7 @@ trait AsyncHandler[F[_]] {
   def acknowledgeCumulativeAsync[T](consumer: api.Consumer[T], messageId: MessageId): F[Unit]
   def acknowledgeCumulativeAsync[T](consumer: api.Consumer[T], messageId: MessageId, txn: Transaction): F[Unit]
   def negativeAcknowledgeAsync[T](consumer: api.Consumer[T], messageId: MessageId): F[Unit]
+  def reconsumeLaterAsync[T](consumer: api.Consumer[T], message: ConsumerMessage[T], delayTime: Long, unit: TimeUnit)(implicit schema: Schema[T]): F[Unit]
 
   def withTransaction[E, A](
     builder: api.transaction.TransactionBuilder,

pulsar4s-core/src/main/scala/com/sksamuel/pulsar4s/Consumer.scala

@@ -1,7 +1,7 @@
 package com.sksamuel.pulsar4s
 
 import com.sksamuel.exts.Logging
-import org.apache.pulsar.client.api.ConsumerStats
+import org.apache.pulsar.client.api.{ConsumerStats, Schema}
 import org.apache.pulsar.client.api.transaction.Transaction
 
 import java.io.Closeable
@@ -94,6 +94,12 @@ trait Consumer[T] extends Closeable with TransactionalConsumerOps[T] {
 
   def negativeAcknowledgeAsync[F[_] : AsyncHandler](messageId: MessageId): F[Unit]
 
+  def reconsumeLater(message: ConsumerMessage[T], delayTime: Long, unit: TimeUnit)
+                    (implicit schema: Schema[T]): Unit
+
+  def reconsumeLaterAsync[F[_] : AsyncHandler](message: ConsumerMessage[T], delayTime: Long, unit: TimeUnit)
+                                              (implicit schema: Schema[T]): F[Unit]
+
   def unsubscribe(): Unit
   def unsubscribeAsync[F[_] : AsyncHandler]: F[Unit]
 
@@ -136,6 +142,14 @@ class DefaultConsumer[T](consumer: JConsumer[T]) extends Consumer[T] with Loggin
   override def negativeAcknowledgeAsync[F[_]: AsyncHandler](messageId: MessageId): F[Unit] =
     implicitly[AsyncHandler[F]].negativeAcknowledgeAsync(consumer, messageId)
 
+  override def reconsumeLater(message: ConsumerMessage[T], delayTime: Long, unit: TimeUnit)
+                             (implicit schema: Schema[T]): Unit =
+    consumer.reconsumeLater(ConsumerMessage.toJava(message, schema), delayTime, unit)
+
+  override def reconsumeLaterAsync[F[_] : AsyncHandler](message: ConsumerMessage[T], delayTime: Long, unit: TimeUnit)
+                                                       (implicit schema: Schema[T]): F[Unit] =
+    implicitly[AsyncHandler[F]].reconsumeLaterAsync(consumer, message, delayTime, unit)
+
   override def stats: ConsumerStats = consumer.getStats
   override def subscription: Subscription = Subscription(consumer.getSubscription)
   override def topic: Topic = Topic(consumer.getTopic)

pulsar4s-core/src/main/scala/com/sksamuel/pulsar4s/ConsumerConfig.scala

@@ -31,5 +31,6 @@ case class ConsumerConfig(subscriptionName: Subscription,
                           additionalProperties: Map[String, AnyRef] = Map.empty,
                           deadLetterPolicy: Option[DeadLetterPolicy] = None,
                           replicateSubscriptionState: Boolean = false,
-                          batchReceivePolicy: Option[BatchReceivePolicy] = None)
+                          batchReceivePolicy: Option[BatchReceivePolicy] = None,
+                          enableRetry: Option[Boolean] = None)
 

pulsar4s-core/src/main/scala/com/sksamuel/pulsar4s/FutureAsyncHandler.scala

@@ -3,9 +3,9 @@ package com.sksamuel.pulsar4s
 import com.sksamuel.pulsar4s.conversions.collections._
 import org.apache.pulsar.client.api
 import org.apache.pulsar.client.api.transaction.Transaction
-import org.apache.pulsar.client.api.{ConsumerBuilder, ReaderBuilder, TypedMessageBuilder}
+import org.apache.pulsar.client.api.{ConsumerBuilder, ReaderBuilder, Schema, TypedMessageBuilder}
 
-import java.util.concurrent.CompletableFuture
+import java.util.concurrent.{CompletableFuture, TimeUnit}
 import scala.compat.java8.FutureConverters
 import scala.compat.java8.FutureConverters.CompletionStageOps
 import scala.concurrent.{ExecutionContext, Future}
@@ -85,6 +85,11 @@ class FutureAsyncHandler(implicit ec: ExecutionContext) extends AsyncHandler[Fut
   override def negativeAcknowledgeAsync[T](consumer: JConsumer[T], messageId: MessageId): Future[Unit] =
     Future.successful(consumer.negativeAcknowledge(messageId))
 
+  override def reconsumeLaterAsync[T](consumer: api.Consumer[T], message: ConsumerMessage[T], delayTime: Long, unit: TimeUnit)
+                                     (implicit schema: Schema[T]): Future[Unit] =
+    Future.successful(consumer.reconsumeLater(ConsumerMessage.toJava(message, schema), delayTime, unit))
+
+
   override def close(reader: api.Reader[_]): Future[Unit] = reader.closeAsync().toScala
 
   override def flush(producer: api.Producer[_]): Future[Unit] = producer.flushAsync().toScala

pulsar4s-core/src/main/scala/com/sksamuel/pulsar4s/PulsarClient.scala

@@ -290,6 +290,7 @@ class DefaultPulsarClient(client: org.apache.pulsar.client.api.PulsarClient) ext
     config.deadLetterPolicy.foreach(builder.deadLetterPolicy)
     config.batchReceivePolicy.foreach(builder.batchReceivePolicy)
     config.acknowledgmentGroupTime.foreach { gt => builder.acknowledgmentGroupTime(gt._1, gt._2) }
+    config.enableRetry.foreach(builder.enableRetry)
     if (config.topics.nonEmpty)
       builder.topics(config.topics.map(_.name).asJava)
     builder.subscriptionName(config.subscriptionName.name)

pulsar4s-monix/src/main/scala/com/sksamuel/pulsar4s/monixs/MonixAsyncHandler.scala

@@ -1,13 +1,12 @@
 package com.sksamuel.pulsar4s.monixs
 
-import java.util.concurrent.CompletableFuture
-
+import java.util.concurrent.{CompletableFuture, TimeUnit}
 import com.sksamuel.pulsar4s
 import com.sksamuel.pulsar4s.conversions.collections._
 import com.sksamuel.pulsar4s.{AsyncHandler, ConsumerMessage, DefaultConsumer, DefaultProducer, DefaultReader, MessageId, Producer, TransactionContext}
 import monix.eval.Task
 import org.apache.pulsar.client.api
-import org.apache.pulsar.client.api.{Consumer, ConsumerBuilder, ProducerBuilder, PulsarClient, Reader, ReaderBuilder, TypedMessageBuilder}
+import org.apache.pulsar.client.api.{Consumer, ConsumerBuilder, ProducerBuilder, PulsarClient, Reader, ReaderBuilder, Schema, TypedMessageBuilder}
 import org.apache.pulsar.client.api.transaction.Transaction
 
 import scala.compat.java8.FutureConverters
@@ -133,6 +132,12 @@ class MonixAsyncHandler extends AsyncHandler[Task] {
 
   override def abortTransaction(txn: Transaction): Task[Unit] =
     Task.deferFuture(txn.abort()).map(_ => ())
+
+  override def reconsumeLaterAsync[T](consumer: Consumer[T], message: ConsumerMessage[T], delayTime: Long, unit: TimeUnit)
+                                     (implicit schema: Schema[T]): Task[Unit] =
+    Task {
+      consumer.reconsumeLater(ConsumerMessage.toJava(message, schema), delayTime, unit)
+    }
 }
 
 object MonixAsyncHandler {

pulsar4s-scalaz/src/main/scala/com/sksamuel/pulsar4s/scalaz/ScalazAsyncHandler.scala

@@ -1,12 +1,11 @@
 package com.sksamuel.pulsar4s.scalaz
 
-import java.util.concurrent.CompletableFuture
-
+import java.util.concurrent.{CompletableFuture, TimeUnit}
 import com.sksamuel.pulsar4s
 import com.sksamuel.pulsar4s.conversions.collections._
 import com.sksamuel.pulsar4s.{AsyncHandler, ConsumerMessage, DefaultConsumer, DefaultProducer, DefaultReader, MessageId, Producer, TransactionContext}
 import org.apache.pulsar.client.api
-import org.apache.pulsar.client.api.{Consumer, ConsumerBuilder, ProducerBuilder, PulsarClient, Reader, ReaderBuilder, TypedMessageBuilder}
+import org.apache.pulsar.client.api.{Consumer, ConsumerBuilder, ProducerBuilder, PulsarClient, Reader, ReaderBuilder, Schema, TypedMessageBuilder}
 import org.apache.pulsar.client.api.transaction.Transaction
 import scalaz.concurrent.Task
 
@@ -128,6 +127,13 @@ class ScalazAsyncHandler extends AsyncHandler[Task] {
   override def commitTransaction(txn: Transaction): Task[Unit] = txn.commit()
 
   override def abortTransaction(txn: Transaction): Task[Unit] = txn.abort()
+
+  override def reconsumeLaterAsync[T](consumer: api.Consumer[T], message: ConsumerMessage[T], delayTime: Long, unit: TimeUnit)
+                                     (implicit schema: Schema[T]): Task[Unit] =
+    Task {
+      consumer.reconsumeLater(ConsumerMessage.toJava(message, schema), delayTime, unit)
+    }
+
 }
 
 object ScalazAsyncHandler {

pulsar4s-zio/src/main/scala/com/sksamuel/pulsar4s/zio/ZioAsyncHandler.scala

@@ -1,12 +1,11 @@
 package com.sksamuel.pulsar4s.zio
 
-import java.util.concurrent.CompletionStage
-
+import java.util.concurrent.{CompletionStage, TimeUnit}
 import com.sksamuel.pulsar4s
 import com.sksamuel.pulsar4s.conversions.collections._
 import com.sksamuel.pulsar4s.{AsyncHandler, ConsumerMessage, DefaultConsumer, DefaultProducer, DefaultReader, MessageId, Producer, TransactionContext}
 import org.apache.pulsar.client.api
-import org.apache.pulsar.client.api.{Consumer, ConsumerBuilder, ProducerBuilder, PulsarClient, Reader, ReaderBuilder, TypedMessageBuilder}
+import org.apache.pulsar.client.api.{Consumer, ConsumerBuilder, ProducerBuilder, PulsarClient, Reader, ReaderBuilder, Schema, TypedMessageBuilder}
 import org.apache.pulsar.client.api.transaction.Transaction
 import zio.{Exit, Task, ZIO}
 
@@ -115,6 +114,10 @@ class ZioAsyncHandler extends AsyncHandler[Task] {
   override def commitTransaction(txn: Transaction): Task[Unit] = fromFuture(ZIO.attempt(txn.commit())).unit
 
   override def abortTransaction(txn: Transaction): Task[Unit] = fromFuture(ZIO.attempt(txn.abort())).unit
+
+  override def reconsumeLaterAsync[T](consumer: Consumer[T], message: ConsumerMessage[T], delayTime: Long, unit: TimeUnit)
+                                     (implicit schema: Schema[T]): Task[Unit] =
+    ZIO.attempt(consumer.reconsumeLater(ConsumerMessage.toJava(message, schema), delayTime, unit))
 }
 
 object ZioAsyncHandler {