This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors-kafka.git
The following commit(s) were added to refs/heads/main by this push:
new 9a317b72 Fix flaky TransactionsSourceSpec: increase idle timeout,
completedWithTimeout threshold, and Await timeout (#553)
9a317b72 is described below
commit 9a317b72583f8093a87ecb7fa225f7bd0278f7b6
Author: PJ Fanning <[email protected]>
AuthorDate: Mon Jun 15 09:51:21 2026 +0100
Fix flaky TransactionsSourceSpec: increase idle timeout,
completedWithTimeout threshold, and Await timeout (#553)
Co-authored-by: copilot-swe-agent[bot]
<[email protected]>
---
.../scala/org/apache/pekko/kafka/TransactionsSourceSpec.scala | 11 +++--------
1 file changed, 3 insertions(+), 8 deletions(-)
diff --git
a/int-tests/src/test/scala/org/apache/pekko/kafka/TransactionsSourceSpec.scala
b/int-tests/src/test/scala/org/apache/pekko/kafka/TransactionsSourceSpec.scala
index d8c02fdf..4d9b6a60 100644
---
a/int-tests/src/test/scala/org/apache/pekko/kafka/TransactionsSourceSpec.scala
+++
b/int-tests/src/test/scala/org/apache/pekko/kafka/TransactionsSourceSpec.scala
@@ -26,7 +26,6 @@ import pekko.stream._
import pekko.stream.scaladsl.{ Flow, Keep, RestartSource, Sink }
import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import org.apache.kafka.clients.producer.ProducerRecord
-import org.scalatest.concurrent.PatienceConfiguration.Interval
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpecLike
@@ -95,12 +94,12 @@ class TransactionsSourceSpec
sourceTopic,
sinkTopic,
transactionId,
- 10.seconds,
+ 30.seconds,
Some(restartAfter),
Some(maxRestarts))
.recover {
case e: TimeoutException =>
- if (completedWithTimeout.incrementAndGet() > 10)
+ if (completedWithTimeout.incrementAndGet() > (consumers *
10))
"no more messages to copy"
else
throw new Error("Continue restarting copy stream")
@@ -120,10 +119,6 @@ class TransactionsSourceSpec
val probeConsumerGroup = createGroupId(2)
- eventually(Interval(2.seconds)) {
- completedCopy.get() should be < consumers
- }
-
val consumer =
offsetValueSource(probeConsumerSettings(probeConsumerGroup), sinkTopic)
.take(elements.toLong)
.alsoTo(
@@ -138,7 +133,7 @@ class TransactionsSourceSpec
.filter(_._2 != "no-more-elements")
.runWith(Sink.seq)
- val values = Await.result(consumer, 10.minutes)
+ val values = Await.result(consumer, 15.minutes)
val expected = (1 to elements).map(_.toString)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
