-
Notifications
You must be signed in to change notification settings - Fork 386
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Travis test debug: repeat flakey tests
- Loading branch information
Showing
12 changed files
with
581 additions
and
256 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
142 changes: 142 additions & 0 deletions
142
tests/src/it/scala/akka/kafka/TransactionsPartitionedSourceSpec.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,142 @@ | ||
| package akka.kafka | ||
|
|
||
| import java.util.concurrent.atomic.AtomicInteger | ||
|
|
||
| import akka.Done | ||
| import akka.kafka.scaladsl.SpecBase | ||
| import akka.kafka.testkit.KafkaTestkitTestcontainersSettings | ||
| import akka.kafka.testkit.scaladsl.TestcontainersKafkaPerClassLike | ||
| import akka.stream._ | ||
| import akka.stream.scaladsl.{Flow, Keep, RestartSource, Sink} | ||
| import akka.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped | ||
| import org.scalatest.concurrent.ScalaFutures | ||
| import org.scalatest.{Matchers, WordSpecLike} | ||
|
|
||
| import scala.collection.immutable | ||
| import scala.concurrent.duration._ | ||
| import scala.concurrent.{Await, Future, TimeoutException} | ||
| import scala.util.{Failure, Success} | ||
|
|
||
| class TransactionsPartitionedSourceSpec extends SpecBase | ||
| with TestcontainersKafkaPerClassLike | ||
| with WordSpecLike | ||
| with ScalaFutures | ||
| with Matchers | ||
| with TransactionsOps | ||
| with Repeated { | ||
|
|
||
| implicit val pc = PatienceConfig(45.seconds, 1.second) | ||
|
|
||
| override val testcontainersSettings = KafkaTestkitTestcontainersSettings(system) | ||
| .withNumBrokers(3) | ||
| .withInternalTopicsReplicationFactor(2) | ||
|
|
||
| "A multi-broker consume-transform-produce cycle" must { | ||
| "provide consistency when multiple partitioned transactional streams are being restarted" in assertAllStagesStopped { | ||
| val sourcePartitions = 10 | ||
| val destinationPartitions = 4 | ||
| val consumers = 3 | ||
| val replication = 2 | ||
|
|
||
| val sourceTopic = createTopic(1, sourcePartitions, replication) | ||
| val sinkTopic = createTopic(2, destinationPartitions, replication) | ||
| val group = createGroupId(1) | ||
|
|
||
| val elements = 100 * 1000 | ||
| val restartAfter = 10 * 1000 / sourcePartitions | ||
|
|
||
| val partitionSize = elements / sourcePartitions | ||
| val producers: immutable.Seq[Future[Done]] = | ||
| (0 until sourcePartitions).map( | ||
| part => produce(sourceTopic, ((part * partitionSize) + 1) to (partitionSize * (part + 1)), part) | ||
| ) | ||
|
|
||
| Await.result(Future.sequence(producers), 1.minute) | ||
|
|
||
| val consumerSettings = consumerDefaults.withGroupId(group) | ||
|
|
||
| val completedCopy = new AtomicInteger(0) | ||
| val completedWithTimeout = new AtomicInteger(0) | ||
|
|
||
| def runStream(id: String): UniqueKillSwitch = | ||
| RestartSource | ||
| .onFailuresWithBackoff(10.millis, 100.millis, 0.2)( | ||
| () => { | ||
| val transactionId = s"$group-$id" | ||
| transactionalPartitionedCopyStream( | ||
| consumerSettings, | ||
| txProducerDefaults, | ||
| sourceTopic, | ||
| sinkTopic, | ||
| transactionId, | ||
| idleTimeout = 10.seconds, | ||
| maxPartitions = sourcePartitions, | ||
| restartAfter = Some(restartAfter) | ||
| ) | ||
| .recover { | ||
| case e: TimeoutException => | ||
| if (completedWithTimeout.incrementAndGet() > 10) | ||
| "no more messages to copy" | ||
| else | ||
| throw new Error("Continue restarting copy stream") | ||
| } | ||
| } | ||
| ) | ||
| .viaMat(KillSwitches.single)(Keep.right) | ||
| .toMat(Sink.onComplete { | ||
| case Success(_) => | ||
| completedCopy.incrementAndGet() | ||
| case Failure(_) => // restart | ||
| })(Keep.left) | ||
| .run() | ||
|
|
||
| val controls: Seq[UniqueKillSwitch] = (0 until consumers) | ||
| .map(_.toString) | ||
| .map(runStream) | ||
|
|
||
| val probeConsumerGroup = createGroupId(2) | ||
|
|
||
| while (completedCopy.get() < consumers) { | ||
| Thread.sleep(2000) | ||
| } | ||
|
|
||
| val consumer = offsetValueSource(probeConsumerSettings(probeConsumerGroup), sinkTopic) | ||
| .take(elements.toLong) | ||
| .idleTimeout(30.seconds) | ||
| .alsoTo( | ||
| Flow[(Long, String)] | ||
| .scan(0) { case (count, _) => count + 1 } | ||
| .filter(_ % 100 == 0) | ||
| .log("received") | ||
| .to(Sink.ignore) | ||
| ) | ||
| .recover { | ||
| case t => (0L, "no-more-elements") | ||
| } | ||
| .filter(_._2 != "no-more-elements") | ||
| .runWith(Sink.seq) | ||
|
|
||
| val values = Await.result(consumer, 10.minutes) | ||
|
|
||
| val expected = (1 to elements).map(_.toString) | ||
|
|
||
| log.debug("Expected elements: {}, actual elements: {}", elements, values.length) | ||
|
|
||
| //println(s"Actual elements:\n$values") | ||
|
|
||
| checkForMissing(values, expected) | ||
| checkForDuplicates(values, expected) | ||
|
|
||
| controls.foreach(_.shutdown()) | ||
| } | ||
| } | ||
|
|
||
| private def probeConsumerSettings(groupId: String): ConsumerSettings[String, String] = | ||
| withProbeConsumerSettings(consumerDefaults, groupId) | ||
|
|
||
| override def producerDefaults: ProducerSettings[String, String] = | ||
| withTestProducerSettings(super.producerDefaults) | ||
|
|
||
| def txProducerDefaults: ProducerSettings[String, String] = | ||
| withTransactionalProducerSettings(super.producerDefaults) | ||
| } |
Oops, something went wrong.