AVRO serialization issue kafkaproducer using scala

user_16473533 0
Viewed 272

I am getting this error, I tried to change the port of schemaregistry from **8081 to 18081. Also I added the dependencies like avro-tools, kafka-schema-registry, etc.

"first_name": "John", "last_name": "Doe", "age": 34, "height": 178.0, "weight": 75.0, "automated_email": false} 21/07/18 19:26:56 INFO clients.Metadata: Cluster ID: Gv4kdRbATLK6YtRxsf94vw Exception in thread "main" org.apache.kafka.common.errors.SerializationException: Error serializing Avro message Caused by: java.net.SocketException: Unexpected end of file from server at java.base/sun.net.www.http.HttpClient.parseHTTPHeader(HttpClient.java:866) at java.base/sun.net.www.http.HttpClient.parseHTTP(HttpClient.java:689) at java.base/sun.net.www.http.HttpClient.parseHTTPHeader(HttpClient.java:863) at java.base/sun.net.www.http.HttpClient.parseHTTP(HttpClient.java:689) at java.base/sun.net.www.protocol.http.HttpURLConnection.getInputStream0(HttpURLConnection.java:1610) at java.base/sun.net.www.protocol.http.HttpURLConnection.getInputStream(HttpURLConnection.java:1515) at java.base/java.net.HttpURLConnection.getResponseCode(HttpURLConnection.java:527) at io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java:212) at io.confluent.kafka.schemaregistry.client.rest.RestService.httpRequest(RestService.java:256) at io.confluent.kafka.schemaregistry.client.rest.RestService.registerSchema(RestService.java:356) at io.confluent.kafka.schemaregistry.client.rest.RestService.registerSchema(RestService.java:348) at io.confluent.kafka.schemaregistry.client.rest.RestService.registerSchema(RestService.java:334) at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.registerAndGetId(CachedSchemaRegistryClient.java:168) at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.register(CachedSchemaRegistryClient.java:222) at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.register(CachedSchemaRegistryClient.java:198) at io.confluent.kafka.serializers.AbstractKafkaAvroSerializer.serializeImpl(AbstractKafkaAvroSerializer.java:70) at io.confluent.kafka.serializers.KafkaAvroSerializer.serialize(KafkaAvroSerializer.java:53) at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:807) at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:784) at com.example.Producerexample$.main(Producerexample.scala:50) at com.example.Producerexample.main(Producerexample.scala)

Process finished with exit code 1

avro-tools confluent-schema-registry
1 Answers

this is the code of above error

package com.example

import java.util.Properties
//import org.apache.avro.util.Utf8
import io.confluent.kafka.serializers.KafkaAvroSerializer
import org.apache.kafka.clients.producer.{KafkaProducer, Producer,ProducerRecord,Callback,RecordMetadata}
import org.apache.kafka.common.serialization.StringSerializer


import com.example.Customer



object Producerexample {


  def main(args: Array[String]): Unit = {
    val properties: Properties = new Properties()
    // normal producer
    properties.setProperty("bootstrap.servers", "http://localhost:9092")
    properties.setProperty("acks", "all")
    properties.setProperty("retries", "10")
    // avro part
    properties.setProperty("key.serializer", classOf[StringSerializer].getName)
    properties.setProperty("value.serializer", classOf[KafkaAvroSerializer].getName)
    properties.setProperty("schema.registry.url", "http://localhost:18081")
    val producer: Producer[String, Customer] =
    new KafkaProducer[String, Customer](properties)
    val topic: String = "customer-avro2"
    // copied from avro examples
     val customer: Customer = Customer.newBuilder()
      .setFirstName("John")
      .setLastName("Doe")
      .setAge(34)
      .setHeight(178f)
      .setWeight(75f)
      .setAutomatedEmail(false)
      .build()
 val producerRecord: ProducerRecord[String, Customer] =
      new ProducerRecord[String, Customer](topic, customer)
    println(customer)
    producer.send(producerRecord, new Callback() {
      def onCompletion(metadata: RecordMetadata,
                       exception: Exception): Unit = {
        if (exception == null) {
          println(metadata)
        } else {
          exception.printStackTrace()
        }
      }
    }
    )
    producer.flush()
    producer.close()
  }
  }
user_164735330
Related