feat: add dead-letter topic support to streaming subscriptions (#1608)

Adds optional deadLetterTopic parameter to the streaming subscribe APIs
on DaprPreviewClient / DaprClientImpl, wiring it through to the
SubscribeTopicEventsRequestInitialAlpha1.dead_letter_topic proto field.

New overloads:
- subscribeToEvents(pubsubName, topic, deadLetterTopic, listener, type)
- subscribeToTopic(pubsubName, topic, deadLetterTopic, type)
- subscribeToTopic(pubsubName, topic, deadLetterTopic, type, metadata)

Existing methods delegate to the new ones with a null deadLetterTopic,
and the field is only set on the request when a non-empty value is
provided. Adds unit tests covering both presence and absence of the
field on the gRPC initial frame.

Fixes: #1608
Signed-off-by: Siri Varma Vegiraju <s_vegiraju@apple.com>

Author: Siri Varma Vegiraju

sdk/src/main/java/io/dapr/client/DaprClientImpl.java

@@ -490,14 +490,29 @@ public <T> Mono<BulkPublishResponse<T>> publishEvents(BulkPublishRequest<T> requ
   @Override
   public <T> Subscription subscribeToEvents(
       String pubsubName, String topic, SubscriptionListener<T> listener, TypeRef<T> type) {
-    DaprPubsubProtos.SubscribeTopicEventsRequestInitialAlpha1 initialRequest =
+    return subscribeToEvents(pubsubName, topic, null, listener, type);
+  }
+
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public <T> Subscription subscribeToEvents(
+      String pubsubName,
+      String topic,
+      String deadLetterTopic,
+      SubscriptionListener<T> listener,
+      TypeRef<T> type) {
+    DaprPubsubProtos.SubscribeTopicEventsRequestInitialAlpha1.Builder initialRequestBuilder =
         DaprPubsubProtos.SubscribeTopicEventsRequestInitialAlpha1.newBuilder()
             .setTopic(topic)
-            .setPubsubName(pubsubName)
-            .build();
+            .setPubsubName(pubsubName);
+    if (deadLetterTopic != null && !deadLetterTopic.isEmpty()) {
+      initialRequestBuilder.setDeadLetterTopic(deadLetterTopic);
+    }
     DaprPubsubProtos.SubscribeTopicEventsRequestAlpha1 request =
         DaprPubsubProtos.SubscribeTopicEventsRequestAlpha1.newBuilder()
-            .setInitialRequest(initialRequest)
+            .setInitialRequest(initialRequestBuilder.build())
             .build();
     return buildSubscription(listener, type, request);
   }
@@ -525,14 +540,35 @@ public <T> Flux<T> subscribeToEvents(String pubsubName, String topic, TypeRef<T>
    */
   @Override
   public <T> Flux<T> subscribeToTopic(String pubsubName, String topic, TypeRef<T> type) {
-    return subscribeToTopic(pubsubName, topic, type, null);
+    return subscribeToTopic(pubsubName, topic, null, type, null);
   }
 
   /**
    * {@inheritDoc}
    */
   @Override
   public <T> Flux<T> subscribeToTopic(String pubsubName, String topic, TypeRef<T> type, Map<String, String> metadata) {
+    return subscribeToTopic(pubsubName, topic, null, type, metadata);
+  }
+
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public <T> Flux<T> subscribeToTopic(String pubsubName, String topic, String deadLetterTopic, TypeRef<T> type) {
+    return subscribeToTopic(pubsubName, topic, deadLetterTopic, type, null);
+  }
+
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public <T> Flux<T> subscribeToTopic(
+      String pubsubName,
+      String topic,
+      String deadLetterTopic,
+      TypeRef<T> type,
+      Map<String, String> metadata) {
     DaprPubsubProtos.SubscribeTopicEventsRequestInitialAlpha1.Builder initialRequestBuilder =
         DaprPubsubProtos.SubscribeTopicEventsRequestInitialAlpha1.newBuilder()
             .setTopic(topic)
@@ -542,6 +578,10 @@ public <T> Flux<T> subscribeToTopic(String pubsubName, String topic, TypeRef<T>
       initialRequestBuilder.putAllMetadata(metadata);
     }
 
+    if (deadLetterTopic != null && !deadLetterTopic.isEmpty()) {
+      initialRequestBuilder.setDeadLetterTopic(deadLetterTopic);
+    }
+
     DaprPubsubProtos.SubscribeTopicEventsRequestAlpha1 request =
         DaprPubsubProtos.SubscribeTopicEventsRequestAlpha1.newBuilder()
             .setInitialRequest(initialRequestBuilder.build())

sdk/src/main/java/io/dapr/client/DaprPreviewClient.java

@@ -292,6 +292,25 @@ <T> Mono<BulkPublishResponse<T>> publishEvents(String pubsubName, String topicNa
   <T> Subscription subscribeToEvents(
       String pubsubName, String topic, SubscriptionListener<T> listener, TypeRef<T> type);
 
+  /**
+   * Subscribe to pubsub via streaming with a dead-letter topic.
+   * @param pubsubName Name of the pubsub component.
+   * @param topic Name of the topic to subscribe to.
+   * @param deadLetterTopic Name of the dead-letter topic to forward failed messages to (null or empty to disable).
+   * @param listener Callback methods to process events.
+   * @param type Type for object deserialization.
+   * @param <T> Type of object deserialization.
+   * @return An active subscription.
+   * @deprecated Use {@link #subscribeToTopic(String, String, String, TypeRef)} instead for a more reactive approach.
+   */
+  @Deprecated
+  <T> Subscription subscribeToEvents(
+      String pubsubName,
+      String topic,
+      String deadLetterTopic,
+      SubscriptionListener<T> listener,
+      TypeRef<T> type);
+
   /**
    * Subscribe to pubsub events via streaming using Project Reactor Flux.
    *
@@ -352,6 +371,36 @@ <T> Subscription subscribeToEvents(
    */
   <T> Flux<T> subscribeToTopic(String pubsubName, String topic, TypeRef<T> type, Map<String, String> metadata);
 
+  /**
+   * Subscribe to pubsub events via streaming using Project Reactor Flux with a dead-letter topic.
+   *
+   * @param pubsubName Name of the pubsub component.
+   * @param topic Name of the topic to subscribe to.
+   * @param deadLetterTopic Name of the dead-letter topic to forward failed messages to (null or empty to disable).
+   * @param type Type for object deserialization.
+   * @return A Flux of deserialized event payloads.
+   * @param <T> Type of the event payload.
+   */
+  <T> Flux<T> subscribeToTopic(String pubsubName, String topic, String deadLetterTopic, TypeRef<T> type);
+
+  /**
+   * Subscribe to pubsub events via streaming using Project Reactor Flux with metadata and dead-letter topic.
+   *
+   * @param pubsubName Name of the pubsub component.
+   * @param topic Name of the topic to subscribe to.
+   * @param deadLetterTopic Name of the dead-letter topic to forward failed messages to (null or empty to disable).
+   * @param type Type for object deserialization.
+   * @param metadata Subscription metadata (e.g., {"rawPayload": "true"}).
+   * @return A Flux of deserialized event payloads.
+   * @param <T> Type of the event payload.
+   */
+  <T> Flux<T> subscribeToTopic(
+      String pubsubName,
+      String topic,
+      String deadLetterTopic,
+      TypeRef<T> type,
+      Map<String, String> metadata);
+
   /*
    * Converse with an LLM.
    *