class ProjectConfig : AbstractProjectConfig() {
    override suspend fun beforeProject(): Unit = TestSystem()
        .with {
            kafka {
                KafkaSystemOptions(
                    configureExposedConfiguration = { cfg ->
                        listOf(
                            "kafka.servers=${cfg.bootstrapServers}",
                            "kafka.interceptor-classes=${cfg.interceptorClass}",
                            "kafka.receive-method=traditional" // here we can change to: 'kotlin-kafka' or 'traditional'
                        )
                    }
                )
            }
            applicationUnderTest(KafkaApplicationUnderTest())
        }.run()

    override suspend fun afterProject(): Unit = TestSystem.stop()
}

class KafkaSystemTests : FunSpec({
  val randomString = { Random.nextInt(0, Int.MAX_VALUE).toString() }

  test("message should be committed and consumed successfully") {
    validate {
      kafka {
        val productId = randomString() + "[productCreated]"
        publish("product", message = ProductCreated(productId), key = randomString().some())
        shouldBeConsumed<ProductCreated>(10.seconds) {
          actual.productId == productId
        }
      }
    }
  }
})