Производитель кафки висит на посылке

Eugen спросил: 13 октября 2017 в 06:34 в: scala

Логика заключается в том, что потоковое задание, получение данных из пользовательского источника, должно записывать как в Kafka, так и в HDFS.

Я написал (очень) базовый производитель Kafka, чтобы сделать это, однако целое потоковое задание зависает в методе отправки.

class KafkaProducer(val kafkaBootstrapServers: String, val kafkaTopic: String, val sslCertificatePath: String, val sslCertificatePassword: String) {  val kafkaProps: Properties = new Properties()
  kafkaProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBootstrapServers)
  kafkaProps.put("acks", "1")
  kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
  kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
  kafkaProps.put("ssl.truststore.location", sslCertificatePath)
  kafkaProps.put("ssl.truststore.password", sslCertificatePassword)  val kafkaProducer: KafkaProducer[Long, Array[String]] = new KafkaProducer(kafkaProps)  def sendKafkaMessage(message: Message): Unit = {
    message.data.foreach(list => {
      val producerRecord: ProducerRecord[Long, Array[String]] = new ProducerRecord[Long, Array[String]](kafkaTopic, message.timeStamp.getTime, list.toArray)
      kafkaProducer.send(producerRecord)
    })
  }
}

И код, вызывающий производителя:

receiverStream.foreachRDD(rdd => {
      val messageRowRDD: RDD[Row] = rdd.mapPartitions(partition => {
        val parser: Parser = new Parser
        val kafkaProducer: KafkaProducer = new KafkaProducer(kafkaBootstrapServers, kafkaTopic, kafkaSslCertificatePath, kafkaSslCertificatePass)
        val newPartition = partition.map(message => {
          Logger.getLogger("importer").error("Writing Message to Kafka...")
          kafkaProducer.sendKafkaMessage(message)
          Logger.getLogger("importer").error("Finished writing Message to Kafka")
          Message.data.map(singleMessage => parser.parseMessage(Message.timeStamp.getTime, singleMessage))
        })
        newPartition.flatten
      })      val df = sqlContext.createDataFrame(messageRowRDD, Schema.messageSchema)      Logger.getLogger("importer").info("Entries-count: " + df.count())
      val row = Try(df.first)      row match {
        case Success(s) => Persister.writeDataframeToDisk(df, outputFolder)
        case Failure(e) => Logger.getLogger("importer").warn("Resulting DataFrame is empty. Nothing can be written")
      }
    })

Из журналов я могу сказать что каждый исполнитель достигает точки "отправки в кафку", но не дальше. Все исполнители вешают это, и никакое исключение не выдается.

Класс Message является очень простым классом case с 2 полями, отметкой времени и массивом строк.

0 ответов