Cloud Stream Dead Letter Topic support #358

ttomsu · 2021-03-10T20:49:33Z

Fixes #158
Two commits here - one for the production code and one for the sample.

Since Cloud Stream handles all things related to topic/subscription creation, we need a way to support dead letter topics. This PR expands the PubSubConsumerProperties to allow setting deadLetterPolicy.deadLetterTopic and deadLetterPolicy.maxDeliveryAttempts

One of the primary changes is consolidating the createSubscription call into its own function which applies the above policy (if available). Because we're potentially creating another topic, I removed the Optional return type from ensureTopicExists and instead have it call itself recursively to ensure we nearly always get a Topic object back and throw an exception in the extremely rare case where we first get an AlreadyExistsException but then still get null back from getTopic(). This is a slight change in behavior because were previously assuming everything was fine and just logging a warning, which we're no longer doing - like I said, though, this should be a pretty rare case.

The tests were refactored to remove the JUnit-4-only @Rule for exception catching, and use AssertJ instead.

The sample code is a combination of the simplicity of the legacy (annotation-based) style and the current functional style, but without the module split out of the other functional sample.

It ends up with 3 functions - one to publish the message, one to receive the message from the main topic, and one to receive the message from the dead letter topic. An integration test ensures this, like all the other tests, starts and runs the application passes as expected.

This PR is already pretty big, so I'll follow up with another PR to pull in some code samples into the refdoc to fix a TODO I put in there earlier.

...n/java/com/google/cloud/spring/stream/binder/pubsub/properties/PubSubConsumerProperties.java

...java/com/google/cloud/spring/stream/binder/pubsub/provisioning/PubSubChannelProvisioner.java

dzou · 2021-03-10T22:51:56Z

...ng-cloud-gcp-pubsub-stream-deadLetterTopic-sample/src/main/java/com/example/SinkExample.java

+public class SinkExample {
+
+	@Bean
+	public Consumer<Message<UserMessage>> logUserMessage() {


Should this also be a Consumer<UserMessage> like the below deadLetterMessages() for consistency? Or does one way only take the wrapped message object?

Unfortunately no, because I need the ability to nack() the message. I've added a utility method that I'd like to get your opinion on for more reliably getting an object I can nack - though I'm not quite sure how this interacts with the acking/nacking happening within the framework. Maybe @elefeint would have some insight.

meltsufin

Very cool! Just some minor comments.

...n/java/com/google/cloud/spring/stream/binder/pubsub/properties/PubSubConsumerProperties.java

...java/com/google/cloud/spring/stream/binder/pubsub/provisioning/PubSubChannelProvisioner.java

meltsufin · 2021-03-11T15:45:38Z

...java/com/google/cloud/spring/stream/binder/pubsub/provisioning/PubSubChannelProvisioner.java

+			DeadLetterPolicy.Builder dlpBuilder = DeadLetterPolicy.newBuilder().setDeadLetterTopic(dlTopic.getName());
+
+			Integer maxAttempts = deadLetterPolicy.getMaxDeliveryAttempts();
+			if (maxAttempts != null && maxAttempts > 0) {


I think I would just let negative values for maxAttempts through. Let it fail there, or maybe there'll be special meaning for negative values.

spring-cloud-gcp-samples/spring-cloud-gcp-pubsub-stream-deadLetterTopic-sample/README.adoc

meltsufin · 2021-03-11T15:55:54Z

spring-cloud-gcp-samples/spring-cloud-gcp-pubsub-stream-deadLetterTopic-sample/README.adoc

+This code sample demonstrates how to use the Spring Cloud Stream binder for Google Cloud Pub/Sub with Dead Letter Topics.
+
+The sample app prompts a user for a message and user name.
+That data is added to a `UserMessage` object, together with the time of message creation, and is sent through Google Cloud Pub/Sub to a sink that `nack()`s the message.


Suggested change

That data is added to a `UserMessage` object, together with the time of message creation, and is sent through Google Cloud Pub/Sub to a sink that `nack()`s the message.

That data is added to a `UserMessage` object, together with the time of message creation, and is sent through Google Cloud Pub/Sub to a sink that `nack()`s the message, which eventually causes the message to be routed to the dead letter topic.

...ng-cloud-gcp-pubsub-stream-deadLetterTopic-sample/src/main/java/com/example/SinkExample.java

meltsufin · 2021-03-11T15:59:35Z

...ng-cloud-gcp-pubsub-stream-deadLetterTopic-sample/src/main/java/com/example/SinkExample.java

+					.orElseThrow(() -> new IllegalStateException("Could not find original PubSubMessage."));
+			Integer deliveryAttempt = Subscriber.getDeliveryAttempt(nackable.getPubsubMessage());
+
+			// Typically you wouldn't nack() every message, but this demonstrates the Pub/Sub system retrying delivery


Doesn't stream automatically ack every message? Are you disabling that?

meltsufin · 2021-03-11T16:00:55Z

...-cloud-gcp-pubsub-stream-deadLetterTopic-sample/src/main/java/com/example/SourceExample.java

+		}
+
+		@Bean
+		Supplier<Flux<UserMessage>> generateUserMessages(Sinks.Many<UserMessage> postOffice) {


Can we do the same without introducing reactive? Would it be simpler?

...java/com/google/cloud/spring/stream/binder/pubsub/provisioning/PubSubChannelProvisioner.java

meltsufin · 2021-03-11T21:27:12Z

spring-cloud-gcp-samples/spring-cloud-gcp-pubsub-stream-dead-letter-sample/README.adoc

+
+```
+Publishing message from batman
+Nacking message 1 from batman at 2021-03-10T15:33:00.479420: to the batcave!


It's going to say (attempt 1) now.

meltsufin · 2021-03-11T21:28:15Z

.../spring-cloud-gcp-pubsub-stream-dead-letter-sample/src/main/resources/application.properties

+spring.cloud.stream.gcp.pubsub.bindings.logUserMessage-in-0.consumer.deadLetterPolicy.maxDeliveryAttempts=6
+# We set this to MANUAL ackMode to take control over the ack()/nack()-ing of messages for this demo.
+# You do not need to do this in a normal application.
+spring.cloud.stream.gcp.pubsub.bindings.logUserMessage-in-0.consumer.ackMode=MANUAL


I'm surprised it worked without this setting for you.

ttomsu · 2021-03-15T15:08:22Z

I'm surprised it worked without this setting for you.

it worked because I nacked the message before the framework could ack it. I'm not sure if Pub/Sub threw any errors for attempting to ack and already nacked message, but it apparently redelivered it all the same.

sonarqubecloud · 2021-03-15T15:15:56Z

Kudos, SonarCloud Quality Gate passed!

0 Bugs
0 Vulnerabilities
0 Security Hotspots
0 Code Smells

94.9% Coverage
0.0% Duplication

* Enable Dead Letter Queue on subscription created by Spring Cloud Stream * Adds dead letter topic sample and integration test

Bumps [checkstyle](https://github.com/checkstyle/checkstyle) from 8.41.1 to 8.42. - [Release notes](https://github.com/checkstyle/checkstyle/releases) - [Commits](checkstyle/checkstyle@checkstyle-8.41.1...checkstyle-8.42) Signed-off-by: dependabot[bot] <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

Travis Tomsu added 2 commits March 10, 2021 15:22

Enable Dead Letter Queue on subscription created by Spring Cloud Stream

6b74853

Adds dead letter topic sample and integration test

4d17f1c

ttomsu requested review from dzou, meltsufin and elefeint March 10, 2021 20:49

dzou reviewed Mar 10, 2021

View reviewed changes

ttomsu mentioned this pull request Mar 11, 2021

Add sample demonstrating the difference between PubSub dead letter queue and Spring Integration error channel #348

Closed

Travis Tomsu added 2 commits March 11, 2021 09:31

dzou comments

7d3213c

checkstyle

85600af

meltsufin requested changes Mar 11, 2021

View reviewed changes

elefeint approved these changes Mar 11, 2021

View reviewed changes

...java/com/google/cloud/spring/stream/binder/pubsub/provisioning/PubSubChannelProvisioner.java Outdated Show resolved Hide resolved

...java/com/google/cloud/spring/stream/binder/pubsub/provisioning/PubSubChannelProvisioner.java Show resolved Hide resolved

Travis Tomsu added 2 commits March 11, 2021 14:15

mike and elena comments

73538a9

ackMode=MANUAL in sample

de6e323

ttomsu requested a review from meltsufin March 11, 2021 19:24

meltsufin approved these changes Mar 11, 2021

View reviewed changes

dzou approved these changes Mar 11, 2021

View reviewed changes

update output in docs

b10d008

ttomsu merged commit 939df3e into main Mar 15, 2021

ttomsu deleted the cloud-stream-dlq branch March 15, 2021 15:18

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Cloud Stream Dead Letter Topic support #358

Cloud Stream Dead Letter Topic support #358

ttomsu commented Mar 10, 2021

dzou Mar 10, 2021

ttomsu Mar 11, 2021

meltsufin left a comment

meltsufin Mar 11, 2021

meltsufin Mar 11, 2021

meltsufin Mar 11, 2021

meltsufin Mar 11, 2021

meltsufin Mar 11, 2021

meltsufin Mar 11, 2021

ttomsu commented Mar 15, 2021

sonarqubecloud bot commented Mar 15, 2021

	That data is added to a `UserMessage` object, together with the time of message creation, and is sent through Google Cloud Pub/Sub to a sink that `nack()`s the message.
	That data is added to a `UserMessage` object, together with the time of message creation, and is sent through Google Cloud Pub/Sub to a sink that `nack()`s the message, which eventually causes the message to be routed to the dead letter topic.

Cloud Stream Dead Letter Topic support #358

Cloud Stream Dead Letter Topic support #358

Conversation

ttomsu commented Mar 10, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

meltsufin left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ttomsu commented Mar 15, 2021

sonarqubecloud bot commented Mar 15, 2021