try to repro

block · Dec 16, 2024 · 5ba487e · 5ba487e
1 parent 889a4da
commit 5ba487e
Show file tree

Hide file tree

Showing 4 changed files with 5 additions and 8 deletions.
diff --git a/backend/runner/pubsub/integration_test.go b/backend/runner/pubsub/integration_test.go
@@ -17,7 +17,7 @@ import (
 )
 
 func TestPubSub(t *testing.T) {
-	t.Skip("Skipping flaky test")
+	// t.Skip("Skipping flaky test")
 	calls := 20
 	events := calls * 10
 	in.Run(t,

diff --git a/backend/runner/pubsub/publisher.go b/backend/runner/pubsub/publisher.go
@@ -10,6 +10,7 @@ import (
 
 	"github.com/block/ftl/backend/timeline"
 	"github.com/block/ftl/common/schema"
+	"github.com/block/ftl/internal/log"
 	"github.com/block/ftl/internal/model"
 	"github.com/block/ftl/internal/rpc"
 )
@@ -76,6 +77,7 @@ func (p *publisher) publish(ctx context.Context, data []byte, key string, caller
 	})
 	if err != nil {
 		timelineEvent.Error = optional.Some(err.Error())
+		log.FromContext(ctx).Errorf(err, "Failed to publish message")
 		return fmt.Errorf("failed to publish message: %w", err)
 	}
 	timelineEvent.Partition = int(partition)

diff --git a/...testdata/java/subscriber/src/main/java/xyz/block/ftl/java/test/subscriber/Subscriber.java b/...testdata/java/subscriber/src/main/java/xyz/block/ftl/java/test/subscriber/Subscriber.java
@@ -12,11 +12,12 @@ public class Subscriber {
 
     @Subscription(topic = TestTopicTopic.class, from = FromOffset.BEGINNING)
     void consume(PubSubEvent event) throws Exception {
-        Log.infof("Subscriber is consuming %s", event.getTime());
+        Log.infof("consume: %s", event.getTime());
     }
 
     @Subscription(topic = TestTopicTopic.class, from = FromOffset.LATEST)
     void consumeFromLatest(PubSubEvent event) throws Exception {
+        Log.infof("consumeFromLatest: %s", event.getTime());
     }
 
     @Subscription(topic = Topic2Topic.class, from = FromOffset.BEGINNING)

diff --git a/...stdata/kotlin/subscriber/src/main/kotlin/xyz/block/ftl/java/test/subscriber/Subscriber.kt b/...stdata/kotlin/subscriber/src/main/kotlin/xyz/block/ftl/java/test/subscriber/Subscriber.kt
@@ -28,12 +28,6 @@ class Subscriber {
         throw RuntimeException("always error: event " + event.time)
     }
 
-    @Subscription(topic = Topic2Topic::class, from = FromOffset.BEGINNING)
-    @Retry(count = 1, minBackoff = "1s", maxBackoff = "1s", catchVerb = "catchAny")
-    fun consumeButFailAndCatchAny(event: PubSubEvent) {
-        throw RuntimeException("always error: event " + event.time)
-    }
-
     @Verb
     @VerbName("catch")
     fun catchVerb(req: CatchRequest<PubSubEvent?>) {