add consuming from dead letter topic to java

backend/runner/pubsub/integration_test.go

@@ -56,7 +56,8 @@ func TestPubSub(t *testing.T) {
 func TestRetry(t *testing.T) {
 	retriesPerCall := 2
 	in.Run(t,
-		in.WithLanguages("java", "go"),
+		// in.WithLanguages("java", "go"),
+		in.WithLanguages("java"),
 
 		in.WithPubSub(),
 		in.CopyModule("publisher"),
@@ -75,7 +76,7 @@ func TestRetry(t *testing.T) {
 		checkPublished("subscriber", "consumeButFailAndRetryFailed", 2),
 
 		// jvm does not allow subscribing to dead letter topics yet
-		in.IfLanguage("go", checkConsumed("subscriber", "consumeFromDeadLetter", true, 2, optional.None[string]())),
+		checkConsumed("subscriber", "consumeFromDeadLetter", true, 2, optional.None[string]()),
 	)
 }
 

backend/runner/pubsub/testdata/java/subscriber/src/main/java/xyz/block/ftl/java/test/subscriber/Subscriber.java

@@ -1,12 +1,22 @@
 package xyz.block.ftl.java.test.subscriber;
 
+import ftl.builtin.FailedEvent;
 import ftl.publisher.PubSubEvent;
 import ftl.publisher.TestTopicTopic;
 import ftl.publisher.Topic2Topic;
 import io.quarkus.logging.Log;
 import xyz.block.ftl.FromOffset;
 import xyz.block.ftl.Retry;
 import xyz.block.ftl.Subscription;
+import xyz.block.ftl.Topic;
+import xyz.block.ftl.TopicPartitionMapper;
+import xyz.block.ftl.WriteableTopic;
+
+class PartitionMapper implements TopicPartitionMapper<PubSubEvent> {
+    public String getPartitionKey(PubSubEvent event) {
+        return event.getTime().toString();
+    }
+}
 
 public class Subscriber {
 
@@ -20,9 +30,20 @@ void consumeFromLatest(PubSubEvent event) throws Exception {
         Log.infof("consumeFromLatest: %s", event.getTime());
     }
 
+    // Java requires the topic to be explicitly defined as an interface for consuming to work
+    @Topic("consumeButFailAndRetryFailed")
+    interface ConsumeButFailAndRetryFailedTopic extends WriteableTopic<PubSubEvent, PartitionMapper> {
+
+    }
+
     @Subscription(topic = Topic2Topic.class, from = FromOffset.BEGINNING, deadLetter = true)
     @Retry(count = 2, minBackoff = "1s", maxBackoff = "1s")
     public void consumeButFailAndRetry(PubSubEvent event) {
         throw new RuntimeException("always error: event " + event.getTime());
     }
+
+    @Subscription(topic = ConsumeButFailAndRetryFailedTopic.class, from = FromOffset.BEGINNING)
+    public void consumeFromDeadLetter(builtin.FailedEvent<PubSubEvent> event) {
+        throw new RuntimeException("always error: event " + event.getEvent().getTime());
+    }
 }