chore: re-enable test thats no longer flaky

block · Dec 17, 2024 · 32b91b0 · 32b91b0
1 parent ca17408
commit 32b91b0
Show file tree

Hide file tree

Showing 4 changed files with 3 additions and 8 deletions.
diff --git a/backend/runner/pubsub/consumer.go b/backend/runner/pubsub/consumer.go
@@ -157,6 +157,7 @@ func (c *consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram
 		for {
 			err := c.call(ctx, msg.Value, int(msg.Partition), int(msg.Offset))
 			if err == nil {
+				logger.Errorf(err, "%s: error consuming message %v:%v", c.verb.Name, msg.Partition, msg.Offset)
 				break
 			}
 			if remainingRetries == 0 {

diff --git a/backend/runner/pubsub/integration_test.go b/backend/runner/pubsub/integration_test.go
@@ -17,7 +17,6 @@ import (
 )
 
 func TestPubSub(t *testing.T) {
-	t.Skip("Skipping flaky test")
 	calls := 20
 	events := calls * 10
 	in.Run(t,

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?>) {