Fix acking behavior of PubSub stream binder (spring-attic#2075)

Fixes a regression in the Pub/Sub stream channel adapter AUTO_ACK mode in which messages were being ACKed even though an exception was thrown.

Author: dzou

spring-cloud-gcp-pubsub/src/main/java/org/springframework/cloud/gcp/pubsub/integration/inbound/PubSubInboundChannelAdapter.java

@@ -145,26 +145,22 @@ public void nack() {
 			});
 		}
 
-		boolean messageNacked = false;
-
 		try {
 			sendMessage(getMessageBuilderFactory()
 					.withPayload(message.getPayload())
 					.copyHeaders(messageHeaders)
 					.build());
+
+			if (this.ackMode == AckMode.AUTO_ACK || this.ackMode == AckMode.AUTO) {
+				message.ack();
+			}
 		}
 		catch (RuntimeException re) {
 			if (this.ackMode == AckMode.AUTO) {
 				message.nack();
-				messageNacked = true;
 			}
 			throw new PubSubException("Sending Spring message failed.", re);
 		}
-		finally {
-			if (!messageNacked && (this.ackMode == AckMode.AUTO || this.ackMode == AckMode.AUTO_ACK)) {
-				message.ack();
-			}
-		}
 	}
 
 }

spring-cloud-gcp-pubsub/src/test/java/org/springframework/cloud/gcp/pubsub/integration/inbound/PubSubInboundChannelAdapterTests.java

@@ -128,6 +128,8 @@ public void testAckModeAutoAck_nacksWhenDownstreamProcessingFails() {
 			this.adapter.start();
 		}).hasMessageContaining(EXCEPTION_MESSAGE);
 
+		// When exception thrown, verify that neither ack() nor nack() is called.
+		verify(mockAcknowledgeableMessage, times(0)).ack();
 		verify(mockAcknowledgeableMessage, times(0)).nack();
 	}