Spring Cloud Stream Solace - Manual Acknowledgement

kirthi

The batch mode is working fine as i am able to receive the msgs as batch and publish as batch.

Now, i am trying to manually acknowledge the messages. If no exception, update the database with success and then accept the message. Else, update the DB with error and then requeue the message.

Function<Message<List<?>>, Collection<Message<?>>> receiveMessage()

return batchMsg -> {

     List<?> batchedPayloads = batchMsg.getPayload();

     System.out.println("size " + batchedPayloads.size());

 

     List<Map<String, Object>> batchedHeaders =

         (List<Map<String, Object>>)

             batchMsg.getHeaders().get(SolaceBinderHeaders.BATCHED_HEADERS);

     AcknowledgmentCallback ackCallback =

         StaticMessageHeaderAccessor.getAcknowledgmentCallback(batchMsg);

     ackCallback.noAutoAck();

//calling the method to do business logic

try { List<> returnObjectList = method();

if (CollectionUtils.isNotEmpty(returnObjectList ))

 {     AckUtils.accept(ackCallback);

       return returnObjectList;    }

   

} catch (Exception e) {

  AckUtils.requeue(ackCallback);

return null;

}

}

application-default.properties

spring.cloud.stream.bindings.receiveMessage-in-0.consumer.max-attempts = 3

spring.cloud.stream.bindings.receiveMessage-in-0.consumer.back-off-initial-interval = 2000

spring.cloud.stream.bindings.receiveMessage-in-0.consumer.back-off-max-interval = 30000

spring.cloud.stream.bindings.receiveMessage-in-0.consumer.back-off-multiplier = 2

spring.cloud.stream.bindings.receiveMessage-in-0.consumer.default-retryable = true

The retry is happening 3 times , but the queue is cleared. Please suggest if the above implementation is fine.

giri

@kirthi Message batches are non-transacted, I believe you cannot manually acknowledge them. I will check on the manual acknowledgment of individual messages and update you.

giri

Hi @kirthi - Just want to confirm that in batch consumers, you can use manual acknowledgment only at the batch level, not at the individual messages.

@Bean
Consumer<Message<List<Payload>>> input() {
	return batchMsg -> { // (1)
            // Disable Auto-Acknowledgement
            AcknowledgmentCallback ackCallback = StaticMessageHeaderAccessor.getAcknowledgmentCallback(batchMsg);
            ackCallback.noAutoAck();

            // manual ack logic
	};
}

Refer to the manual acknowledgment has details on manual ack logic.

kirthi

@giri Thanks for replying.

So when user defined exception occurs, if we reject the message. will all messages in the batch will be in solace queue and will be retried from queue?

giri

@kirthi That is correct - should confess, I have not tested this.