fix: removing delivery attempt attribute when dead lettering is not enabled (#72)

* fix: creating fix to not populate delivery attempt attribute when dead lettering is not enabled * Adding unit test for case in which a received message has no delivery attempt

Author: hannahrogers-google

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java

@@ -346,10 +346,16 @@ private void processBatch(List<OutstandingMessage> batch) {
  }
 
  private PubsubMessage addDeliveryInfoCount(ReceivedMessage receivedMessage) {
- return PubsubMessage.newBuilder(receivedMessage.getMessage())
- .putAttributes(
- "googclient_deliveryattempt", Integer.toString(receivedMessage.getDeliveryAttempt()))
- .build();
+ PubsubMessage originalMessage = receivedMessage.getMessage();
+ int deliveryAttempt = receivedMessage.getDeliveryAttempt();
+ // Delivery Attempt will be set to 0 if DeadLetterPolicy is not set on the subscription. In
+ // this case, do not populate the PubsubMessage with the delivery attempt attribute.
+ if (deliveryAttempt > 0) {
+ return PubsubMessage.newBuilder(originalMessage)
+ .putAttributes("googclient_deliveryattempt", Integer.toString(deliveryAttempt))
+ .build();
+ }
+ return originalMessage;
  }
 
  private void processOutstandingMessage(final PubsubMessage message, final AckHandler ackHandler) {

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Subscriber.java

@@ -207,8 +207,11 @@ public static Builder newBuilder(String subscription, MessageReceiver receiver)
  }
 
  /** Returns the delivery attempt count for a received {@link PubsubMessage} */
- public static int getDeliveryAttempt(PubsubMessage message) {
- return Integer.parseInt(message.getAttributesOrDefault("googclient_deliveryattempt", "0"));
+ public static Integer getDeliveryAttempt(PubsubMessage message) {
+ if (!message.containsAttributes("googclient_deliveryattempt")) {
+ return null;
+ }
+ return Integer.parseInt(message.getAttributesOrThrow("googclient_deliveryattempt"));
  }
 
  /** Subscription which the subscriber is subscribed to. */

google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/MessageDispatcherTest.java

@@ -17,6 +17,8 @@
 package com.google.cloud.pubsub.v1;
 
 import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 
 import com.google.api.gax.batching.FlowControlSettings;
 import com.google.api.gax.batching.FlowController;
@@ -59,6 +61,7 @@ public void run() {
  private List<ModAckItem> sentModAcks;
  private FakeClock clock;
  private FlowController flowController;
+ private boolean messageContainsDeliveryAttempt;
 
  @AutoValue
  abstract static class ModAckItem {
@@ -82,8 +85,13 @@ public void setUp() {
  @Override
  public void receiveMessage(final PubsubMessage message, final AckReplyConsumer consumer) {
  assertThat(message.getData()).isEqualTo(MESSAGE_DATA);
- assertThat(message.getAttributesOrThrow("googclient_deliveryattempt"))
- .isEqualTo(Integer.toString(DELIVERY_INFO_COUNT));
+ if (messageContainsDeliveryAttempt) {
+ assertTrue(message.containsAttributes("googclient_deliveryattempt"));
+ assertThat(message.getAttributesOrThrow("googclient_deliveryattempt"))
+ .isEqualTo(Integer.toString(DELIVERY_INFO_COUNT));
+ } else {
+ assertFalse(message.containsAttributes("googclient_deliveryattempt"));
+ }
  consumers.add(consumer);
  }
  };
@@ -126,6 +134,8 @@ public void sendAckOperations(
  systemExecutor,
  clock);
  dispatcher.setMessageDeadlineSeconds(Subscriber.MIN_ACK_DEADLINE_SECONDS);
+
+ messageContainsDeliveryAttempt = true;
  }
 
  @Test
@@ -136,6 +146,22 @@ public void testReceipt() {
  .contains(ModAckItem.of(TEST_MESSAGE.getAckId(), Subscriber.MIN_ACK_DEADLINE_SECONDS));
  }
 
+ @Test
+ public void testReceiptNoDeliveryAttempt() {
+ messageContainsDeliveryAttempt = false;
+ ReceivedMessage messageNoDeliveryAttempt =
+ ReceivedMessage.newBuilder()
+ .setAckId("ackid")
+ .setMessage(PubsubMessage.newBuilder().setData(MESSAGE_DATA).build())
+ .build();
+ dispatcher.processReceivedMessages(Collections.singletonList(messageNoDeliveryAttempt));
+ dispatcher.processOutstandingAckOperations();
+ assertThat(sentModAcks)
+ .contains(
+ ModAckItem.of(
+ messageNoDeliveryAttempt.getAckId(), Subscriber.MIN_ACK_DEADLINE_SECONDS));
+ }
+
  @Test
  public void testAck() throws Exception {
  dispatcher.processReceivedMessages(Collections.singletonList(TEST_MESSAGE));

google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/SubscriberTest.java

@@ -86,15 +86,16 @@ public void tearDown() throws Exception {
 
  @Test
  public void testDeliveryAttemptHelper() {
- int deliveryAttempt = 3;
+ Integer deliveryAttempt = 3;
  PubsubMessage message =
  PubsubMessage.newBuilder()
  .putAttributes("googclient_deliveryattempt", Integer.toString(deliveryAttempt))
  .build();
  assertEquals(Subscriber.getDeliveryAttempt(message), deliveryAttempt);
 
+ // In the case where delivery attempt attribute is not populated, expect null
  PubsubMessage emptyMessage = PubsubMessage.newBuilder().build();
- assertEquals(Subscriber.getDeliveryAttempt(emptyMessage), 0);
+ assertEquals(Subscriber.getDeliveryAttempt(emptyMessage), null);
  }
 
  @Test