All batch is redelivered even if some messages are acked

[KannarFr]

When a consumer receives a batch, when some messages are nacked and others acked, all the batch is redelivered.

It seems related to PIP-54. After reading pulsar java client, ConsumerConfigurationData as a batchIndexAckEnabled which has default to false.

I think we all need to support it in pulsar4s ConsumerConfig.

@gmethvin I saw you created the first issue: apache/pulsar#5969, WDYT?

[KannarFr]

I have not specified that I use batch, is batch enabled by default?

[gmethvin]

Batching is enabled on the producer by default, but batch index acknowledgement on the consumer is not. I assume that's the case mainly for backwards compatibility reasons. You need to enable it by setting enableBatchIndexAcknowledgment on the ConsumerBuilder, but currently there is no equivalent option in ConsumerConfig, so that would need to be added.

As a workaround, you can create the consumer directly using the Java client:

val pulsar4sConsumer: Future[pulsar4s.Consumer[T]] =
  javaClient
    .newConsumer(implicitly[Schema[T]])
    .enableBatchIndexAcknowledgement(true)
    .subscribeAsync()
    .asScala
    .map(new pulsar4s.DefaultConsumer(_))

[KannarFr]

Ok, thanks for the workaround!

[gmethvin]

I'm closing this because it's an issue with the underlying Java client, not pulsar4s. It probably doesn't make sense for pulsar4s to have different default behavior than the Java client.