import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.{StringDeserializer, StringSerializer}

object SparkKafkaValidation {

  def main(args: Array[String]): Unit = {
    // Создаем конфигурацию Spark
    val sparkConf = new SparkConf().setAppName("SparkKafkaValidation").setMaster("local[2]")
    val ssc = new StreamingContext(sparkConf, Seconds(5))

    // Настройки для Kafka
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "localhost:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "spark-consumer-group",
      "auto.offset.reset" -> "earliest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    // Топики Kafka
    val topics = Array("input-topic")

    // Создаем DStream, принимающий сообщения из Kafka
    val stream = KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
    )

    // Функция для валидации сообщения
    def validateMessage(message: String): Boolean = {
      message.length > 5
    }

    // Отправка сообщения в топик Kafka
    def sendToKafka(topic: String, message: String): Unit = {
      val producerProperties = new java.util.Properties()
      producerProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
      producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getName)
      producerProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getName)

      val producer = new KafkaProducer[String, String](producerProperties)
      producer.send(new ProducerRecord[String, String](topic, message))
      producer.close()
    }

    // Обрабатываем каждое сообщение из Kafka
    stream.foreachRDD { rdd =>
      rdd.foreach { record =>
        val message = record.value()
        if (validateMessage(message)) {
          // Отправляем валидное сообщение в топик Kafka
          sendToKafka("valid-topic", message)
          println(s"Valid message: $message")
        } else {
          // Отправляем ошибку в топик Kafka
          sendToKafka("error-topic", s"Invalid message: $message")
          println(s"Invalid message: $message")
        }
      }
    }

    // Запускаем Spark Streaming
    ssc.start()
    ssc.awaitTermination()
  }
}