Kafka - Unable to Send a Message to a Remote Server Using Java

Kafka - Unable to send a message to a remote server using Java

In your kafka server.properties there is a commented configuration

#advertised.host.name=<Some IP>

Uncomment this and add the IP of the Linux Machine in which kafka is running.

advertised.host.name=<Kafka Running Machine IP>

And connect from clients to <Kafka Running Machine IP> This should fix your issue.

EDIT

Optionally you can uncomment the

#advertised.port=9092

Also if you are listening on a different port than the default one.

Scala : Unable to send message to Kafka (hosted on remote server)

Please update your code as below and try once. It looks like you have not closed the output stream,encoder and producer properly.

val producer = new KafkaProducer[String, Array[Byte]](configuration)
  val writer = new SpecificDatumWriter[GenericRecord](schema)
  val out = new ByteArrayOutputStream()
  val encoder = EncoderFactory.get.binaryEncoder(out, null)
  writer.write(avroRecord, encoder)

  val serializedBytes: Array[Byte] = out.toByteArray()

  encoder.flush()
  out.close()

  val recordToSend = new ProducerRecord[String, Array[Byte]](topic, serializedBytes)
  producer.send(recordToSend,new ProducerCallback)

  //flush data
  producer.flush()
  //flush and close producer
  producer.close()

class ProducerCallback(implicit logger: Logger) extends Callback {

  override def onCompletion(metadata: RecordMetadata, exception: Exception): Unit = {
    //executes every time a record is successfully sent or exception thrown
    Option(metadata) match {
      case Some(_) =>
        logger.info("Received new metadata. \n" +
          "Topic: " + metadata.topic() + "\n" +
          "Partition: " + metadata.partition() + "\n" +
          "Offset: " + metadata.offset() + "\n" +
          "Timestamp: " + metadata.timestamp() + "\n" +
          "Checksum: " + metadata.checksum())
      case None => ;
    }
    Option(exception) match {
      case Some(_) =>
        logger.error("Exception thrown during processing of record... " + exception)
        throw exception
      case None => ;
    }
  }
}

Please refer link https://github.com/Zapagol/apache-kafka/tree/master/src/main/scala/com/org/apache for more kafka producer and consumer examples. Hope it will help!

Update

I have added KafkaProducer example for Avroschema input. Please refer https://github.com/Zapagol/apache-kafka/blob/master/src/main/scala/com/org/apache/producers/ProducerForAvroschema.scala .

I have used apache avro jar and sample avsc file as below. Please modify schema file according to your requirement.And I am able to produce record successfully.

{
   "type": "record",
   "name": "employee",
   "fields": [
      {"name": "name", "type": "string"},
      {"name": "id", "type": "int"},
      {"name": "mobileNumber", "type": ["string", "null"]},
      {"name": "salary", "type": ["int", "null"]}
  ]
}

Not able consume messages from remote machine in Kafka

By default the broker will bind to localhost. If your machine ip is a.b.c.d and it's a vm instance, then you need to uncomment the Line in server.properties containing listeners=PLAINTEXT://:9092 and put listeners=PLAINTEXT://a.b.c.d:9092

If it's a docker container you can try adding the following two lines:

listeners=PLAINTEXT://0.0.0.0:9092
advertised.listeners=PLAINTEXT://a.b.c.d:9092

Kafka messages not accepted by server (only from remote IP?)

Found. Based on Kafka - Unable to send a message to a remote server using Java, the problem was on the config file conf/servers.properties; it requires uncommenting this line:

advertised.listeners=PLAINTEXT://192.168.1.131:9092

Thanks anyway.

Not able to communicate from a remote machine to Kafka cluster

The issue was solved.

The problem was due to one of the host in the cluster had conflict in configuration, that is why messages were not being through. After updating the configs it worked like charm.

Kafka: java client failed to send messages after x tries

Apache kafka has a new producer client that is better:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.8.2.0</version>
</dependency>

Example:
https://github.com/CameronGregory/kafka/blob/master/TestProducer.java

Aparently your config is ok. Is "test.StringEncoder" your custom class? try yo use "kafka.serializer.StringEncoder" instead

---

Related Topics