Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Fix flaky tests #7223

Merged
merged 6 commits into from
Nov 27, 2024
Merged

Fix flaky tests #7223

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
40fce5e
add test start logger
mslabek Nov 22, 2024
085d803
logging in flaky test
mslabek Nov 26, 2024
b1521c0
try to fix periodic test
mslabek Nov 26, 2024
cac0d67
try to fix kafka test not waiting for created topic
mslabek Nov 26, 2024
21bad8d
revert temp changes
mslabek Nov 26, 2024
d57db70
rename
mslabek Nov 27, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 8 additions & 5 deletions ...l/touk/nussknacker/engine/management/periodic/PeriodicProcessServiceIntegrationTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import org.scalatest.concurrent.ScalaFutures
import org.scalatest.exceptions.TestFailedException
import org.scalatest.flatspec.AnyFlatSpec
import org.scalatest.matchers.should.Matchers
import org.scalatest.time.{Millis, Seconds, Span}
import org.testcontainers.utility.DockerImageName
import pl.touk.nussknacker.engine.api.deployment.simple.SimpleStateStatus
import pl.touk.nussknacker.engine.api.deployment.simple.SimpleStateStatus.ProblemStateStatus
Expand Down Expand Up @@ -428,9 +429,12 @@ class PeriodicProcessServiceIntegrationTest
toDeployAfterFinish should have length 1
toDeployAfterFinish.head.scheduleName.value.value shouldBe secondSchedule

val activities = service.getScenarioActivitiesSpecificToPeriodicProcess(processIdWithName).futureValue
val firstActivity = activities.head.asInstanceOf[ScenarioActivity.PerformedScheduledExecution]
activities shouldBe List(
val firstActivity = eventually {
val result = service.getScenarioActivitiesSpecificToPeriodicProcess(processIdWithName).futureValue
result should not be empty
result.head.asInstanceOf[ScenarioActivity.PerformedScheduledExecution]
}
firstActivity shouldBe
ScenarioActivity.PerformedScheduledExecution(
scenarioId = ScenarioId(1),
scenarioActivityId = firstActivity.scenarioActivityId,
Expand All @@ -443,8 +447,7 @@ class PeriodicProcessServiceIntegrationTest
createdAt = firstActivity.createdAt,
retriesLeft = None,
nextRetryAt = None
),
)
)
}

it should "handle multiple one time schedules" in withFixture() { f =>
Expand Down
16 changes: 9 additions & 7 deletions utils/kafka-test-utils/src/main/scala/pl/touk/nussknacker/engine/kafka/KafkaClient.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,13 +7,12 @@ import org.apache.kafka.clients.admin.{NewTopic, TopicDescription}
import org.apache.kafka.clients.consumer.KafkaConsumer
import org.apache.kafka.clients.producer.{Callback, KafkaProducer, ProducerRecord, RecordMetadata}
import org.apache.kafka.common.header.Headers
import retry.When

import java.time.Duration
import java.util
import java.util.{Collections, UUID}
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.{Future, Promise}
import scala.concurrent.{Await, Future, Promise}
import scala.jdk.CollectionConverters._
import scala.util.{Failure, Success, Try}
import scala.concurrent.duration._
Expand All @@ -33,12 +32,15 @@ class KafkaClient(kafkaAddress: String, id: String) extends LazyLogging {
def createTopic(name: String, partitions: Int = 5): Unit = {
adminClient.createTopics(Collections.singletonList(new NewTopic(name, partitions, 1: Short))).all().get()
// When kraft enabled, topics doesn't appear instantly after createTopic
retry.Pause(10, 1.second)(Timer.default)(
Future {
topic(name)
}
val timeout = 10.seconds
Await.result(
retry.Pause(10, 1.second)(Timer.default)(
Future {
topic(name)
}
),
timeout
)

}

def deleteTopic(name: String): Unit =
Expand Down
Loading