Интеграционный тест Flink и Kafka с scalatest-embedded-kafka

Question

Я хотел бы запустить интеграционный тест с Flink и Kafka. Процесс состоит в том, чтобы прочитать из Кафки некоторые манипуляции с Флинком и поместить поток данных в Кафку.

Я хотел бы проверить процесс от начала до конца. Пока я использую https://github.com/manub/scalatest-embedded-kafka.

Я привел пример, который я постарался сделать максимально простым:

import java.util.Properties

import net.manub.embeddedkafka.{EmbeddedKafka, EmbeddedKafkaConfig}
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.streaming.api.functions.sink.SinkFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer011, FlinkKafkaProducer011}
import org.scalatest.{Matchers, WordSpec}

import scala.collection.mutable.ListBuffer

object SimpleFlinkKafkaTest {

  class CollectSink extends SinkFunction[String] {
    override def invoke(string: String): Unit = {
      synchronized {
        CollectSink.values += string
      }
    }
  }

  object CollectSink {
    val values: ListBuffer[String] = ListBuffer.empty[String]
  }

  val kafkaPort = 9092
  val zooKeeperPort = 2181

  val props = new Properties()
  props.put("bootstrap.servers", "localhost:" + kafkaPort.toString)
  props.put("schema.registry.url", "localhost:" + zooKeeperPort.toString)

  val inputString = "mystring"
  val expectedString = "MYSTRING"
}

class SimpleFlinkKafkaTest extends WordSpec with Matchers with EmbeddedKafka {

  "runs with embedded kafka" should {

    "work" in {

      implicit val config = EmbeddedKafkaConfig(
        kafkaPort = SimpleFlinkKafkaTest.kafkaPort,
        zooKeeperPort = SimpleFlinkKafkaTest.zooKeeperPort
      )

      withRunningKafka {

        publishStringMessageToKafka("input-topic", SimpleFlinkKafkaTest.inputString)

        val env = StreamExecutionEnvironment.getExecutionEnvironment

        env.setParallelism(1)

        val kafkaConsumer = new FlinkKafkaConsumer011(
          "input-topic",
          new SimpleStringSchema,
          SimpleFlinkKafkaTest.props
        )

        implicit val typeInfo = TypeInformation.of(classOf[String])

        val inputStream = env.addSource(kafkaConsumer)

        val outputStream = inputStream.map(_.toUpperCase)

        val kafkaProducer = new FlinkKafkaProducer011(
          "output-topic",
          new SimpleStringSchema(),
          SimpleFlinkKafkaTest.props
        )
        outputStream.addSink(kafkaProducer)
        env.execute()
        consumeFirstStringMessageFrom("output-topic") shouldEqual SimpleFlinkKafkaTest.expectedString

      }
    }
  }
}

У меня была ошибка, поэтому я добавляю строку implicit val typeInfo = TypeInformation.of(classOf[String]) но я не очень понимаю, почему я должен это делать.

Пока этот код не работает, он работает без прерываний, но не останавливается и не дает никакого результата.

Если у кого-то есть идеи? Еще лучшая идея для тестирования этого вида трубопровода.

Спасибо!

РЕДАКТИРОВАТЬ: добавить env.execute() и поменяй ошибку.

Answer 1

Вот простое решение, которое я придумал.

Идея состоит в том, чтобы:

1. Запустите Kafka Embedded server
2. Создайте свои тестовые темы (здесь ввод и вывод)
3. Запустите задание Flink в будущем, чтобы не блокировать основной поток
4. Опубликовать сообщение в теме ввода
5. Проверьте результат в теме вывода

И рабочий прототип:

import java.util.Properties

import org.apache.flink.streaming.api.scala._
import net.manub.embeddedkafka.{EmbeddedKafka, EmbeddedKafkaConfig}
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.core.fs.FileSystem.WriteMode
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer011, FlinkKafkaProducer011}
import org.scalatest.{Matchers, WordSpec}

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future

class SimpleFlinkKafkaTest extends WordSpec with Matchers with EmbeddedKafka {

    "runs with embedded kafka on arbitrary available ports" should {

        val env = StreamExecutionEnvironment.getExecutionEnvironment

        "work" in {
            val userDefinedConfig = EmbeddedKafkaConfig(kafkaPort = 9092, zooKeeperPort = 2182)

            val properties = new Properties()
            properties.setProperty("bootstrap.servers", "localhost:9092")
            properties.setProperty("zookeeper.connect", "localhost:2182")
            properties.setProperty("group.id", "test")
            properties.setProperty("auto.offset.reset", "earliest")

            val kafkaConsumer = new FlinkKafkaConsumer011[String]("input", new SimpleStringSchema(), properties)
            val kafkaSink = new FlinkKafkaProducer011[String]("output", new SimpleStringSchema(), properties)
            val stream = env
                .addSource(kafkaConsumer)
                .map(_.toUpperCase)
                .addSink(kafkaSink)

            withRunningKafkaOnFoundPort(userDefinedConfig) { implicit actualConfig =>
                createCustomTopic("input")
                createCustomTopic("output")
                Future{env.execute()}
                publishStringMessageToKafka("input", "Titi")
                consumeFirstStringMessageFrom("output") shouldEqual "TITI"
            }
        }
    }
}