SaFi Bank Space : Slacker Retry Implementation

Current:

slacker manager will subscribe a subscription under google topic
send a ack message to google pubsub after process the message successfully
send a nack message to google pubsub if got an exception, google will resend the same message again to slacker once it received the nack message. this will repeat unlimited times.

~~if the network breaks or the message exceed the deadline expiration time , it will keep staying in message queue and never resend~~

Future

don’t need to implement any retry feature anymore
send ack/nack to google pub
if network break or message exceed deadline expire time, the message will move to dlq topics

Enable DLQ message in Google topic

turn on the dlq features in google Pub/Sub, then the message will move to dlq topic once it exceed the deadline expire time

Keep error information in message

the google pub/sub messages can’t be updated once it published, so in this scenario, we need to clone message from original and add error log information in the clone message and send it to DLQ topic

Pseudo-code

fun onMessage(message: PubsubMessage, acknowledgement: Acknowledgement) {
  
    val atttempt_count = message.attributesMap.get("googclient_deliveryattempt")
    logger.info("Received messageId: ${message.messageId}, content: ${message.data.toStringUtf8()}, attempt count: $atttempt_count" )
    try {
      process message
    }catch (e : Exception){
      if (atttempt_count != null) {
          if(Integer.parseInt(atttempt_count) <= 3){
              acknowledgement.nack()
          }else{
              logger.info("Send message to DLQ")
              acknowledgement.ack()
          }
      }
    }
}

Logs：

17:18:23.696 [pubsub-scheduled-executor-thread-1] INFO  p.s.s.m.GooglePubSubCommandListener - Received messageId: 6040212615481823, content: 
        {
           "commandType":"DEBUG_MESSAGE",
           "customerId":"2a237743-4ea0-3d9e-bc35-b89b78d39011",
           "params":{
              "message": "2022-10-24 17:18:22"
           }
        }
        , attempt count: 1
17:18:30.915 [pubsub-scheduled-executor-thread-1] INFO  p.s.s.m.GooglePubSubCommandListener - Received messageId: 6040212615481823, content: 
        {
           "commandType":"DEBUG_MESSAGE",
           "customerId":"2a237743-4ea0-3d9e-bc35-b89b78d39011",
           "params":{
              "message": "2022-10-24 17:18:22"
           }
        }
        , attempt count: 2
17:18:33.239 [pubsub-scheduled-executor-thread-1] INFO  p.s.s.m.GooglePubSubCommandListener - Received messageId: 6040212615481823, content: 
        {
           "commandType":"DEBUG_MESSAGE",
           "customerId":"2a237743-4ea0-3d9e-bc35-b89b78d39011",
           "params":{
              "message": "2022-10-24 17:18:22"
           }
        }
        , attempt count: 3
17:18:33.644 [pubsub-scheduled-executor-thread-1] INFO  p.s.s.m.GooglePubSubCommandListener - Received messageId: 6040212615481823, content: 
        {
           "commandType":"DEBUG_MESSAGE",
           "customerId":"2a237743-4ea0-3d9e-bc35-b89b78d39011",
           "params":{
              "message": "2022-10-24 17:18:22"
           }
        }
        , attempt count: 4
17:18:33.644 [pubsub-scheduled-executor-thread-1] INFO  p.s.s.m.GooglePubSubCommandListener - Send message to DLQ

NACK Block message POC

POC Code

 val slackerCommand = mapper.readValue(message.data.toStringUtf8(), SlackerCommand::class.java)
        val customerId = slackerCommand.customerId
        if( customerId.toString() == "00000000-0000-0000-0000-000000000000"){
            if (atttempt_count != null) {
                if(Integer.parseInt(atttempt_count) <= 3){
                    logger.info("reject message: ${message.messageId}, content: ${message.data.toStringUtf8()}")
                    acknowledgement.nack()
                }else{
                    logger.info("Send message to DLQ")
                    acknowledgement.ack()
                }
            }
        }else{
            logger.info("approve message: ${message.messageId}, content: ${message.data.toStringUtf8()}")
            acknowledgement.ack()
        }

Messages in google pubs:

block message which customer id = “00000000-0000-0000-0000-000000000000“ and accept other messages.
check if the message processor will be blocked

Conclusion:

Google send message in random order, if the message blocked the processor , it will not block other process. We will move this blocked message to dlq queue after retry 3 times and then send acknowledgement to tell google pub skip this message and process next for blocked message. please check below logs.

17:26:54.618 [pubsub-scheduled-executor-thread-1] INFO  p.s.s.m.GooglePubSubCommandListener - Received messageId: 6055838669968923, content: {
  "commandType": "SEND_NOTIFICATION",
  "customerId": "2b827c7b-a7e9-4116-af78-940963fd85e4",
  "params": {
      "messageType": "RISK_TRANSACTION_REJECTED",
      "amount": "1102"
   }
}, attempt count: 1
17:26:55.051 [pubsub-scheduled-executor-thread-1] INFO  p.s.s.m.GooglePubSubCommandListener - approve message: 6055838669968923, content: {
  "commandType": "SEND_NOTIFICATION",
  "customerId": "2b827c7b-a7e9-4116-af78-940963fd85e4",
  "params": {
      "messageType": "RISK_TRANSACTION_REJECTED",
      "amount": "1102"
   }
}
17:26:55.054 [pubsub-scheduled-executor-thread-1] INFO  p.s.s.m.GooglePubSubCommandListener - Received messageId: 6055838670030697, content: {
  "commandType": "SEND_NOTIFICATION",
  "customerId": "2b827c7b-a7e9-4116-af78-940963fd85e4",
  "params": {
      "messageType": "RISK_TRANSACTION_REJECTED",
      "amount": "1101"
   }
}, attempt count: 1
17:26:55.054 [pubsub-scheduled-executor-thread-1] INFO  p.s.s.m.GooglePubSubCommandListener - approve message: 6055838670030697, content: {
  "commandType": "SEND_NOTIFICATION",
  "customerId": "2b827c7b-a7e9-4116-af78-940963fd85e4",
  "params": {
      "messageType": "RISK_TRANSACTION_REJECTED",
      "amount": "1101"
   }
}
17:26:55.054 [pubsub-scheduled-executor-thread-1] INFO  p.s.s.m.GooglePubSubCommandListener - Received messageId: 6055838670001673, content: {
  "commandType": "SEND_NOTIFICATION",
  "customerId": "00000000-0000-0000-0000-000000000000",
  "params": {
      "messageType": "RISK_TRANSACTION_REJECTED",
      "amount": "1100"
   }
}, attempt count: 1
17:26:55.055 [pubsub-scheduled-executor-thread-1] INFO  p.s.s.m.GooglePubSubCommandListener - reject message: 6055838670001673, content: {
  "commandType": "SEND_NOTIFICATION",
  "customerId": "00000000-0000-0000-0000-000000000000",
  "params": {
      "messageType": "RISK_TRANSACTION_REJECTED",
      "amount": "1100"
   }
}
17:26:55.425 [pubsub-scheduled-executor-thread-1] INFO  p.s.s.m.GooglePubSubCommandListener - Received messageId: 6055838670001673, content: {
  "commandType": "SEND_NOTIFICATION",
  "customerId": "00000000-0000-0000-0000-000000000000",
  "params": {
      "messageType": "RISK_TRANSACTION_REJECTED",
      "amount": "1100"
   }
}, attempt count: 2
17:26:55.425 [pubsub-scheduled-executor-thread-1] INFO  p.s.s.m.GooglePubSubCommandListener - reject message: 6055838670001673, content: {
  "commandType": "SEND_NOTIFICATION",
  "customerId": "00000000-0000-0000-0000-000000000000",
  "params": {
      "messageType": "RISK_TRANSACTION_REJECTED",
      "amount": "1100"
   }
}
17:26:55.833 [pubsub-scheduled-executor-thread-1] INFO  p.s.s.m.GooglePubSubCommandListener - Received messageId: 6055838670001673, content: {
  "commandType": "SEND_NOTIFICATION",
  "customerId": "00000000-0000-0000-0000-000000000000",
  "params": {
      "messageType": "RISK_TRANSACTION_REJECTED",
      "amount": "1100"
   }
}, attempt count: 3
17:26:55.833 [pubsub-scheduled-executor-thread-1] INFO  p.s.s.m.GooglePubSubCommandListener - reject message: 6055838670001673, content: {
  "commandType": "SEND_NOTIFICATION",
  "customerId": "00000000-0000-0000-0000-000000000000",
  "params": {
      "messageType": "RISK_TRANSACTION_REJECTED",
      "amount": "1100"
   }
}
17:26:56.234 [pubsub-scheduled-executor-thread-1] INFO  p.s.s.m.GooglePubSubCommandListener - Received messageId: 6055838670001673, content: {
  "commandType": "SEND_NOTIFICATION",
  "customerId": "00000000-0000-0000-0000-000000000000",
  "params": {
      "messageType": "RISK_TRANSACTION_REJECTED",
      "amount": "1100"
   }
}, attempt count: 4
17:26:56.234 [pubsub-scheduled-executor-thread-1] INFO  p.s.s.m.GooglePubSubCommandListener - Send message to DLQ
17:28:43.606 [pubsub-scheduled-executor-thread-1] INFO  p.s.s.m.GooglePubSubCommandListener - Received messageId: 6055890800229636, content: {
  "commandType": "SEND_NOTIFICATION",
  "customerId": "2b827c7b-a7e9-4116-af78-940963fd85e4",
  "params": {
      "messageType": "RISK_TRANSACTION_REJECTED",
      "amount": "1104"
   }
}, attempt count: 1
17:28:43.608 [pubsub-scheduled-executor-thread-1] INFO  p.s.s.m.GooglePubSubCommandListener - approve message: 6055890800229636, content: {
  "commandType": "SEND_NOTIFICATION",
  "customerId": "2b827c7b-a7e9-4116-af78-940963fd85e4",
  "params": {
      "messageType": "RISK_TRANSACTION_REJECTED",
      "amount": "1104"
   }
}