Spring Cloud Stream Solace - Manual Acknowledgement

kirthi
kirthi Member Posts: 22 ✭✭
edited March 2023 in Connectors & Integrations #1

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.

Comments

  • giri
    giri Member, Administrator, Employee Posts: 116 admin

    @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
    giri Member, Administrator, Employee Posts: 116 admin

    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
    kirthi Member Posts: 22 ✭✭

    @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
    giri Member, Administrator, Employee Posts: 116 admin

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