diff --git a/examples/src/main/java/io/dapr/examples/pubsub/stream/README.md b/examples/src/main/java/io/dapr/examples/pubsub/stream/README.md
index 0f10980988..969138738e 100644
--- a/examples/src/main/java/io/dapr/examples/pubsub/stream/README.md
+++ b/examples/src/main/java/io/dapr/examples/pubsub/stream/README.md
@@ -108,6 +108,38 @@ client.subscribeToTopic(PUBSUB_NAME, topicName, TypeRef.STRING, Map.of("rawPaylo
     .blockLast();
 ```
 
+### Subscription with a Dead-Letter Topic
+
+For workloads that need to forward unprocessable messages to a dead-letter topic,
+use the listener-based `subscribeToEvents` overload that accepts a
+`deadLetterTopic` argument. When the listener returns `Status.DROP`, the Dapr
+runtime publishes the message to the configured dead-letter topic:
+
+```java
+var listener = new SubscriptionListener<String>() {
+  @Override
+  public Mono<Status> onEvent(CloudEvent<String> event) {
+    if (shouldRejectMessage(event)) {
+      return Mono.just(Status.DROP); // forwarded to deadLetterTopicName
+    }
+    return Mono.just(Status.SUCCESS);
+  }
+
+  @Override
+  public void onError(RuntimeException exception) {
+    System.out.println("Subscriber got exception: " + exception.getMessage());
+  }
+};
+
+try (var subscription = client.subscribeToEvents(
+        PUBSUB_NAME, topicName, deadLetterTopicName, listener, TypeRef.STRING)) {
+  subscription.awaitTermination();
+}
+```
+
+See [SubscriberWithDeadLetter.java](SubscriberWithDeadLetter.java) for a complete example
+that consumes the dead-letter topic in the same process.
+
 ### Subscription Lifecycle
 
 The examples use `blockLast()` to keep the subscriber running indefinitely. For production use cases requiring explicit subscription lifecycle control, you can use `.subscribe()` which returns a `Disposable` that can be disposed via `disposable.dispose()`.
@@ -138,6 +170,13 @@ Or run the CloudEvent Subscriber example:
 dapr run --resources-path ./components/pubsub --app-id subscriber -- java -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.pubsub.stream.SubscriberCloudEvent
 ```
 
+Or run the dead-letter Subscriber example, which routes messages whose payload
+contains "fail" to a dead-letter topic and consumes both topics:
+
+```bash
+dapr run --resources-path ./components/pubsub --app-id subscriber -- java -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.pubsub.stream.SubscriberWithDeadLetter
+```
+
 Once the subscriber is running, run the publisher in a new terminal to see the events in the subscriber's side:
 
 <!-- STEP
diff --git a/examples/src/main/java/io/dapr/examples/pubsub/stream/SubscriberWithDeadLetter.java b/examples/src/main/java/io/dapr/examples/pubsub/stream/SubscriberWithDeadLetter.java
new file mode 100644
index 0000000000..8fcf91bdb2
--- /dev/null
+++ b/examples/src/main/java/io/dapr/examples/pubsub/stream/SubscriberWithDeadLetter.java
@@ -0,0 +1,104 @@
+/*
+ * Copyright 2024 The Dapr Authors
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package io.dapr.examples.pubsub.stream;
+
+import io.dapr.client.DaprClientBuilder;
+import io.dapr.client.SubscriptionListener;
+import io.dapr.client.domain.CloudEvent;
+import io.dapr.utils.TypeRef;
+import reactor.core.publisher.Mono;
+
+/**
+ * Subscriber using bi-directional gRPC streaming with a dead-letter topic.
+ *
+ * <p>This example demonstrates how to forward failed messages to a dead-letter topic.
+ * The main listener returns {@code Status.DROP} for messages whose payload contains
+ * "fail", which causes the Dapr runtime to publish them to the configured
+ * dead-letter topic. A second subscription consumes the dead-letter topic so the
+ * forwarded messages are visible in the same process.
+ *
+ * <p>Usage:
+ * <ol>
+ *   <li>Build and install jars: {@code mvn clean install}
+ *   <li>cd [repo root]/examples
+ *   <li>Run the subscriber:
+ *       {@code dapr run --resources-path ./components/pubsub --app-id subscriber -- \
+ *         java -jar target/dapr-java-sdk-examples-exec.jar \
+ *         io.dapr.examples.pubsub.stream.SubscriberWithDeadLetter}
+ *   <li>Publish messages from another terminal, e.g. containing the word "fail" to
+ *       see them routed to the dead-letter topic.
+ * </ol>
+ */
+public class SubscriberWithDeadLetter {
+
+  private static final String DEFAULT_TOPIC_NAME = "testingtopic";
+  private static final String DEFAULT_DEAD_LETTER_TOPIC_NAME = "testingtopic-deadletter";
+  private static final String PUBSUB_NAME = "messagebus";
+
+  /**
+   * Main entry point for the dead-letter subscriber example.
+   *
+   * @param args Optional positional args: [topicName] [deadLetterTopicName].
+   * @throws Exception An Exception on startup.
+   */
+  public static void main(String[] args) throws Exception {
+    String topicName = args.length >= 1 ? args[0] : DEFAULT_TOPIC_NAME;
+    String deadLetterTopicName = args.length >= 2 ? args[1] : DEFAULT_DEAD_LETTER_TOPIC_NAME;
+
+    try (var client = new DaprClientBuilder().buildPreviewClient()) {
+      System.out.println("Subscribing to dead-letter topic: " + deadLetterTopicName);
+
+      var deadLetterListener = new SubscriptionListener<String>() {
+        @Override
+        public Mono<Status> onEvent(CloudEvent<String> event) {
+          System.out.println("Dead-letter subscriber got: " + event.getData());
+          return Mono.just(Status.SUCCESS);
+        }
+
+        @Override
+        public void onError(RuntimeException exception) {
+          System.out.println("Dead-letter subscriber got exception: " + exception.getMessage());
+        }
+      };
+
+      System.out.println("Subscribing to topic: " + topicName
+          + " (dead-letter topic: " + deadLetterTopicName + ")");
+
+      var mainListener = new SubscriptionListener<String>() {
+        @Override
+        public Mono<Status> onEvent(CloudEvent<String> event) {
+          String data = event.getData();
+          if (data != null && data.toLowerCase().contains("fail")) {
+            System.out.println("Subscriber dropping message to dead-letter: " + data);
+            return Mono.just(Status.DROP);
+          }
+          System.out.println("Subscriber got: " + data);
+          return Mono.just(Status.SUCCESS);
+        }
+
+        @Override
+        public void onError(RuntimeException exception) {
+          System.out.println("Subscriber got exception: " + exception.getMessage());
+        }
+      };
+
+      try (var deadLetterSubscription = client.subscribeToEvents(
+              PUBSUB_NAME, deadLetterTopicName, deadLetterListener, TypeRef.STRING);
+           var mainSubscription = client.subscribeToEvents(
+              PUBSUB_NAME, topicName, deadLetterTopicName, mainListener, TypeRef.STRING)) {
+        mainSubscription.awaitTermination();
+      }
+    }
+  }
+}
diff --git a/sdk-tests/src/test/java/io/dapr/it/pubsub/stream/PubSubStreamIT.java b/sdk-tests/src/test/java/io/dapr/it/pubsub/stream/PubSubStreamIT.java
index 334bc5232a..9b0b78ef2f 100644
--- a/sdk-tests/src/test/java/io/dapr/it/pubsub/stream/PubSubStreamIT.java
+++ b/sdk-tests/src/test/java/io/dapr/it/pubsub/stream/PubSubStreamIT.java
@@ -46,6 +46,8 @@ public class PubSubStreamIT extends BaseIT {
   private static final String TOPIC_NAME_FLUX = "stream-topic-flux";
   private static final String TOPIC_NAME_CLOUDEVENT = "stream-topic-cloudevent";
   private static final String TOPIC_NAME_RAWPAYLOAD = "stream-topic-rawpayload";
+  private static final String TOPIC_NAME_DLQ = "stream-topic-dlq";
+  private static final String TOPIC_NAME_DLQ_DEADLETTER = "stream-topic-dlq-deadletter";
   private static final String PUBSUB_NAME = "messagebus";
 
   private final List<DaprRun> runs = new ArrayList<>();
@@ -261,4 +263,77 @@ public void testPubSubRawPayload() throws Exception {
       disposable.dispose();
     }
   }
+
+  @Test
+  public void testPubSubDeadLetterTopic() throws Exception {
+    final DaprRun daprRun = closeLater(startDaprApp(
+        this.getClass().getSimpleName() + "-dlq",
+        60000));
+
+    var runId = UUID.randomUUID().toString();
+    try (DaprClient client = daprRun.newDaprClient();
+         DaprPreviewClient previewClient = daprRun.newDaprPreviewClient()) {
+
+      // Subscribe to the dead-letter topic first so we don't miss any messages.
+      Set<String> deadLetterMessageIds = Collections.synchronizedSet(new HashSet<>());
+      var deadLetterListener = new SubscriptionListener<String>() {
+        @Override
+        public Mono<Status> onEvent(CloudEvent<String> event) {
+          if (event.getData() != null && event.getData().contains(runId)) {
+            deadLetterMessageIds.add(event.getId());
+            System.out.println("Received dead-letter message ID: " + event.getId());
+          }
+          return Mono.just(Status.SUCCESS);
+        }
+
+        @Override
+        public void onError(RuntimeException exception) {
+          System.err.println("Dead-letter subscription error: " + exception.getMessage());
+        }
+      };
+
+      // Subscribe to the main topic with a listener that always DROPs, which should
+      // forward the messages to the dead-letter topic.
+      var mainListener = new SubscriptionListener<String>() {
+        @Override
+        public Mono<Status> onEvent(CloudEvent<String> event) {
+          if (event.getData() != null && event.getData().contains(runId)) {
+            System.out.println("Dropping message ID: " + event.getId());
+            return Mono.just(Status.DROP);
+          }
+          return Mono.just(Status.DROP);
+        }
+
+        @Override
+        public void onError(RuntimeException exception) {
+          System.err.println("Main subscription error: " + exception.getMessage());
+        }
+      };
+
+      try (var deadLetterSubscription = previewClient.subscribeToEvents(
+              PUBSUB_NAME, TOPIC_NAME_DLQ_DEADLETTER, deadLetterListener, TypeRef.STRING);
+           var mainSubscription = previewClient.subscribeToEvents(
+              PUBSUB_NAME, TOPIC_NAME_DLQ, TOPIC_NAME_DLQ_DEADLETTER, mainListener, TypeRef.STRING)) {
+
+        // Publish messages to the main topic.
+        for (int i = 0; i < NUM_MESSAGES; i++) {
+          String message = String.format("DLQ message #%d for run %s", i, runId);
+          client.publishEvent(PUBSUB_NAME, TOPIC_NAME_DLQ, message).block();
+        }
+
+        callWithRetry(() -> {
+          var count = deadLetterMessageIds.size();
+          System.out.println(
+              String.format("Got %d dead-letter messages out of %d for topic %s.",
+                  count, NUM_MESSAGES, TOPIC_NAME_DLQ_DEADLETTER));
+          assertEquals(NUM_MESSAGES, deadLetterMessageIds.size());
+        }, 120000);
+
+        mainSubscription.close();
+        mainSubscription.awaitTermination();
+        deadLetterSubscription.close();
+        deadLetterSubscription.awaitTermination();
+      }
+    }
+  }
 }
diff --git a/sdk/src/main/java/io/dapr/client/DaprClientImpl.java b/sdk/src/main/java/io/dapr/client/DaprClientImpl.java
index e5c95ec0d7..3adb3bc37f 100644
--- a/sdk/src/main/java/io/dapr/client/DaprClientImpl.java
+++ b/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,7 +540,7 @@ 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);
   }
 
   /**
@@ -533,6 +548,27 @@ public <T> Flux<T> subscribeToTopic(String pubsubName, String topic, TypeRef<T>
    */
   @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())
diff --git a/sdk/src/main/java/io/dapr/client/DaprPreviewClient.java b/sdk/src/main/java/io/dapr/client/DaprPreviewClient.java
index bb5ae5e1fa..d0d2e3ea66 100644
--- a/sdk/src/main/java/io/dapr/client/DaprPreviewClient.java
+++ b/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.
    *
diff --git a/sdk/src/test/java/io/dapr/client/DaprPreviewClientGrpcTest.java b/sdk/src/test/java/io/dapr/client/DaprPreviewClientGrpcTest.java
index ab82896dd1..b8d1f3f6b2 100644
--- a/sdk/src/test/java/io/dapr/client/DaprPreviewClientGrpcTest.java
+++ b/sdk/src/test/java/io/dapr/client/DaprPreviewClientGrpcTest.java
@@ -95,6 +95,7 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -948,6 +949,253 @@ public void onCompleted() {
     assertEquals("true", capturedMetadata.get().get("rawPayload"));
   }
 
+  @Test
+  public void subscribeToEventsWithDeadLetterTopicTest() throws Exception {
+    var pubsubName = "pubsubName";
+    var topicName = "topicName";
+    var deadLetterTopic = "topicName-DLQ";
+    var started = new Semaphore(0);
+    var capturedInitial = new AtomicReference<DaprPubsubProtos.SubscribeTopicEventsRequestInitialAlpha1>();
+    var gotInitial = new Semaphore(0);
+
+    doAnswer((Answer<StreamObserver<DaprPubsubProtos.SubscribeTopicEventsRequestAlpha1>>) invocation -> {
+      StreamObserver<DaprPubsubProtos.SubscribeTopicEventsResponseAlpha1> observer =
+          (StreamObserver<DaprPubsubProtos.SubscribeTopicEventsResponseAlpha1>) invocation.getArguments()[0];
+
+      new Thread(() -> {
+        try {
+          started.acquire();
+        } catch (InterruptedException e) {
+          throw new RuntimeException(e);
+        }
+        observer.onNext(DaprPubsubProtos.SubscribeTopicEventsResponseAlpha1.getDefaultInstance());
+        observer.onCompleted();
+      }).start();
+
+      return new StreamObserver<>() {
+        @Override
+        public void onNext(DaprPubsubProtos.SubscribeTopicEventsRequestAlpha1 request) {
+          if (request.hasInitialRequest()) {
+            capturedInitial.set(request.getInitialRequest());
+            gotInitial.release();
+          }
+          started.release();
+        }
+
+        @Override
+        public void onError(Throwable throwable) {
+        }
+
+        @Override
+        public void onCompleted() {
+        }
+      };
+    }).when(daprStub).subscribeTopicEventsAlpha1(any(StreamObserver.class));
+
+    var subscription = previewClient.subscribeToEvents(
+        pubsubName,
+        topicName,
+        deadLetterTopic,
+        new SubscriptionListener<String>() {
+          @Override
+          public Mono<Status> onEvent(CloudEvent<String> event) {
+            return Mono.just(Status.SUCCESS);
+          }
+
+          @Override
+          public void onError(RuntimeException exception) {
+          }
+        },
+        TypeRef.STRING);
+
+    gotInitial.acquire();
+    subscription.close();
+
+    assertNotNull(capturedInitial.get());
+    assertEquals(pubsubName, capturedInitial.get().getPubsubName());
+    assertEquals(topicName, capturedInitial.get().getTopic());
+    assertTrue(capturedInitial.get().hasDeadLetterTopic());
+    assertEquals(deadLetterTopic, capturedInitial.get().getDeadLetterTopic());
+  }
+
+  @Test
+  public void subscribeToEventsWithoutDeadLetterTopicDoesNotSetFieldTest() throws Exception {
+    var pubsubName = "pubsubName";
+    var topicName = "topicName";
+    var started = new Semaphore(0);
+    var capturedInitial = new AtomicReference<DaprPubsubProtos.SubscribeTopicEventsRequestInitialAlpha1>();
+    var gotInitial = new Semaphore(0);
+
+    doAnswer((Answer<StreamObserver<DaprPubsubProtos.SubscribeTopicEventsRequestAlpha1>>) invocation -> {
+      StreamObserver<DaprPubsubProtos.SubscribeTopicEventsResponseAlpha1> observer =
+          (StreamObserver<DaprPubsubProtos.SubscribeTopicEventsResponseAlpha1>) invocation.getArguments()[0];
+
+      new Thread(() -> {
+        try {
+          started.acquire();
+        } catch (InterruptedException e) {
+          throw new RuntimeException(e);
+        }
+        observer.onNext(DaprPubsubProtos.SubscribeTopicEventsResponseAlpha1.getDefaultInstance());
+        observer.onCompleted();
+      }).start();
+
+      return new StreamObserver<>() {
+        @Override
+        public void onNext(DaprPubsubProtos.SubscribeTopicEventsRequestAlpha1 request) {
+          if (request.hasInitialRequest()) {
+            capturedInitial.set(request.getInitialRequest());
+            gotInitial.release();
+          }
+          started.release();
+        }
+
+        @Override
+        public void onError(Throwable throwable) {
+        }
+
+        @Override
+        public void onCompleted() {
+        }
+      };
+    }).when(daprStub).subscribeTopicEventsAlpha1(any(StreamObserver.class));
+
+    var subscription = previewClient.subscribeToEvents(
+        pubsubName,
+        topicName,
+        new SubscriptionListener<String>() {
+          @Override
+          public Mono<Status> onEvent(CloudEvent<String> event) {
+            return Mono.just(Status.SUCCESS);
+          }
+
+          @Override
+          public void onError(RuntimeException exception) {
+          }
+        },
+        TypeRef.STRING);
+
+    gotInitial.acquire();
+    subscription.close();
+
+    assertNotNull(capturedInitial.get());
+    assertFalse(capturedInitial.get().hasDeadLetterTopic());
+  }
+
+  @Test
+  public void subscribeToTopicWithDeadLetterTopicTest() throws Exception {
+    var pubsubName = "pubsubName";
+    var topicName = "topicName";
+    var deadLetterTopic = "topicName-DLQ";
+    var started = new Semaphore(0);
+    var capturedInitial = new AtomicReference<DaprPubsubProtos.SubscribeTopicEventsRequestInitialAlpha1>();
+
+    doAnswer((Answer<StreamObserver<DaprPubsubProtos.SubscribeTopicEventsRequestAlpha1>>) invocation -> {
+      StreamObserver<DaprPubsubProtos.SubscribeTopicEventsResponseAlpha1> observer =
+          (StreamObserver<DaprPubsubProtos.SubscribeTopicEventsResponseAlpha1>) invocation.getArguments()[0];
+
+      new Thread(() -> {
+        try {
+          started.acquire();
+        } catch (InterruptedException e) {
+          throw new RuntimeException(e);
+        }
+        observer.onNext(DaprPubsubProtos.SubscribeTopicEventsResponseAlpha1.getDefaultInstance());
+        observer.onCompleted();
+      }).start();
+
+      return new StreamObserver<>() {
+        @Override
+        public void onNext(DaprPubsubProtos.SubscribeTopicEventsRequestAlpha1 request) {
+          if (request.hasInitialRequest()) {
+            capturedInitial.set(request.getInitialRequest());
+          }
+          started.release();
+        }
+
+        @Override
+        public void onError(Throwable throwable) {
+        }
+
+        @Override
+        public void onCompleted() {
+        }
+      };
+    }).when(daprStub).subscribeTopicEventsAlpha1(any(StreamObserver.class));
+
+    var disposable = previewClient.subscribeToTopic(pubsubName, topicName, deadLetterTopic, TypeRef.STRING)
+        .subscribe();
+
+    // Wait briefly for the initial request to be captured, then dispose.
+    for (int i = 0; i < 50 && capturedInitial.get() == null; i++) {
+      Thread.sleep(20);
+    }
+    disposable.dispose();
+
+    assertNotNull(capturedInitial.get());
+    assertEquals(pubsubName, capturedInitial.get().getPubsubName());
+    assertEquals(topicName, capturedInitial.get().getTopic());
+    assertTrue(capturedInitial.get().hasDeadLetterTopic());
+    assertEquals(deadLetterTopic, capturedInitial.get().getDeadLetterTopic());
+  }
+
+  @Test
+  public void subscribeToTopicWithDeadLetterTopicAndMetadataTest() throws Exception {
+    var pubsubName = "pubsubName";
+    var topicName = "topicName";
+    var deadLetterTopic = "topicName-DLQ";
+    var started = new Semaphore(0);
+    var capturedInitial = new AtomicReference<DaprPubsubProtos.SubscribeTopicEventsRequestInitialAlpha1>();
+
+    doAnswer((Answer<StreamObserver<DaprPubsubProtos.SubscribeTopicEventsRequestAlpha1>>) invocation -> {
+      StreamObserver<DaprPubsubProtos.SubscribeTopicEventsResponseAlpha1> observer =
+          (StreamObserver<DaprPubsubProtos.SubscribeTopicEventsResponseAlpha1>) invocation.getArguments()[0];
+
+      new Thread(() -> {
+        try {
+          started.acquire();
+        } catch (InterruptedException e) {
+          throw new RuntimeException(e);
+        }
+        observer.onNext(DaprPubsubProtos.SubscribeTopicEventsResponseAlpha1.getDefaultInstance());
+        observer.onCompleted();
+      }).start();
+
+      return new StreamObserver<>() {
+        @Override
+        public void onNext(DaprPubsubProtos.SubscribeTopicEventsRequestAlpha1 request) {
+          if (request.hasInitialRequest()) {
+            capturedInitial.set(request.getInitialRequest());
+          }
+          started.release();
+        }
+
+        @Override
+        public void onError(Throwable throwable) {
+        }
+
+        @Override
+        public void onCompleted() {
+        }
+      };
+    }).when(daprStub).subscribeTopicEventsAlpha1(any(StreamObserver.class));
+
+    Map<String, String> metadata = Map.of("rawPayload", "true");
+    var disposable = previewClient
+        .subscribeToTopic(pubsubName, topicName, deadLetterTopic, TypeRef.STRING, metadata)
+        .subscribe();
+
+    for (int i = 0; i < 50 && capturedInitial.get() == null; i++) {
+      Thread.sleep(20);
+    }
+    disposable.dispose();
+
+    assertNotNull(capturedInitial.get());
+    assertTrue(capturedInitial.get().hasDeadLetterTopic());
+    assertEquals(deadLetterTopic, capturedInitial.get().getDeadLetterTopic());
+    assertEquals("true", capturedInitial.get().getMetadataMap().get("rawPayload"));
+  }
+
   @Test
   public void converseShouldThrowIllegalArgumentExceptionWhenComponentNameIsNull() throws Exception {
     List<ConversationInput> inputs = new ArrayList<>();