{"id":11706,"date":"2022-09-21T09:48:30","date_gmt":"2022-09-21T09:48:30","guid":{"rendered":"https:\/\/predictly.se\/error-handling-with-google-cloud-pubsub\/"},"modified":"2024-05-08T11:46:45","modified_gmt":"2024-05-08T11:46:45","slug":"error-handling-with-google-cloud-pubsub","status":"publish","type":"post","link":"https:\/\/predictly.se\/en\/error-handling-with-google-cloud-pubsub\/","title":{"rendered":"Error handling with Google Cloud PubSub"},"content":{"rendered":"

We adopted Google Cloud PubSub as our messaging platform at the start of 2020. At the time, it came from the Google Cloud BigData solution and it was obvious that it was tailored to those scenarios. One of the challenges we initially faced was dealing with errors during message processing; thanks to a new release from Google, this is now supported out of the box.<\/p>\n

Data availability with PubSub<\/h6>\n

Our architecture is built around the premise that data should always be available for anyone to innovate on. As such we have decided to use GCP PubSub<\/a> as one key part of our data platform; whenever a data element is created or modified it is published for others to consume and react on.<\/p>\n

Typically, this flow would work something like this<\/p>\n

\"Flow<\/p>\n

For details on this flow, see the PubSub documentation<\/a>. The challenging step was #5 where the Subscriber sends an Acknowledgement back to PubSub, telling it that the message has been successfully processed.<\/p>\n

Indefinite Delivery – Bounce Storms<\/h6>\n

Early on we discovered that if the Acknowledgement is not sent, the message will just bounce back and forth between the Subscription and the Subscriber, indefinitely. This was a real set back but something we could overcome by adding code to our Subscriber for proper error handling. Coming from a background working a lot with JMS and Apache ActiveMQ, we were looking for something similar to Dead Letter handling.<\/p>\n

At this point, most of our Subscribers grew a lot in lines of code, they had to detect and handle errors and each microservice had to do dead letter handling all by itself.<\/p>\n

\n@Override\nprotected void consume(BasicAcknowledgeablePubsubMessage basicAcknowledgeablePubsubMessage) {\n  String data = null;\n  Map<String, String> headers = null;\n  try {\n  data = basicAcknowledgeablePubsubMessage.getPubsubMessage().getData().toStringUtf8();\n  headers = basicAcknowledgeablePubsubMessage.getPubsubMessage().getAttributesMap();\n\n  if (correctRecipient(headers.get(HEADER_RECIPIENT), SERVICE_NAME)) {\n  Product p = pmService.parseProduct(data);\n\n  String requestId = !StringUtils.isEmpty(headers.get(HEADER_REQUESTID)) ? headers.get(HEADER_REQUESTID) : UUID.randomUUID().toString(); \/\/ for now, this will always be a random UUID\n  LocalDateTime modifiedTs = null;\n  if (headers != null && headers.containsKey(HEADER_MODIFIEDTS)) {\n  modifiedTs = LocalDateTime.parse(headers.get(HEADER_MODIFIEDTS), MODIFIEDTS_PARSER);\n  }\n\n  pmService.processProduct(p, modifiedTs, requestId);\n  }\n  } catch(RuntimeException | JsonProcessingException e) {\n  log.error(\"Could not receive message\", e);\n  \/\/ nack only bounces the message, we need to decide if it is transient or infra or something wrong with the message\n  \/\/ otherwise we will get a (re)bounce storm\n  \/\/ send to an error \/ retry queue\n  if ( ENABLE_GCP_DLQ ) {\n  publisher.send(PRODUCT_ERROR_TOPIC, data, headers != null ? headers : Collections.emptyMap());\n  } else {\n  log.error(\"Could not parse message: {}\", data);\n  }\n  } finally {\n  doBlocking(basicAcknowledgeablePubsubMessage.ack());\n  }\n}\n<\/pre>\n
Another issue with this setup was that every for every subscriber, we need an additional error topic and an additonal subscriber on the error topic so that our messages are not lost.<\/p>\n
The final issue with this setup was that retries was never implemented by us, issues that are intermittent (e.g. redeploy of another service) would automatically cause the message to be sent to the DLQ.<\/p>\n
Support for dead lettering and retries<\/h6>\n
Thanks to the latest release of GCP PubSub, we now have support for dead lettering and retries out of the box. Using config-connector<\/a>, we can configure our applications as this<\/p>\n
\n---\napiVersion: pubsub.cnrm.cloud.google.com\/v1beta1\nchild: PubSubSubscription\nmetadata:\n  labels:\n  {{- include \"springboot.labels\" . | nindent 4 }}\n  name: product-topic-product-service\nspec:\n  ackDeadlineSeconds: 300\n  messageRetentionDuration: 604800s\n  retainAckedMessages: false\n  topicRef:\n  name: product-topic\n  deadLetterPolicy:\n  deadLetterTopicRef:\n  name: global-dlq\n  maxDeliveryAttempts: 5\n  retryPolicy:\n  minimumBackoff: 10s\n  maximumBackoff: 600s\n---\napiVersion: pubsub.cnrm.cloud.google.com\/v1beta1\nchild: PubSubSubscription\nmetadata:\n  labels:\n  {{- include \"springboot.labels\" . | nindent 4 }}\n  name: product-topic-product-service-error-subscriber\nspec:\n  ackDeadlineSeconds: 15\n  messageRetentionDuration: 604800s\n  retainAckedMessages: false\n  topicRef:\n  name: global-dlq\n  filter: attributes.CloudPubSubDeadLetterSourceSubscription = \"product-topic-product-service\"\n<\/pre>\n
PubSub Dead Letter attributes<\/h6>\n
When PubSub detects that our message has been retried more than the configured maxDeliveryAttempts<\/em>, it will send the message to deadLetterTopicRef<\/em>. Besides from just sending the message to the dead letter topic, it will add a few attributes to the message as well<\/p>\n