Not able to start the kafka broker

[embeddedkafka]

Not able to start the kafka broker

[francescopellegrini]

Hi @tharunrajguptayada, could you please provide a snippet of your code?

[tharunrajguptayada]

yeah sure , i am attaching my code below
`class AvroMockTesting extends AnyWordSpecLike with Matchers with BeforeAndAfterAll {

val schema: Schema = new Schema.Parser().parse("{\n "type": "record",\n // "namespace": "com.example",\n "name": "Customer",\n "version": "1",\n "fields": [\n { "name": "first_name", "type": "string", "doc": "First Name of Customer" },\n { "name": "automated_email", "type": "boolean", "default": true, "doc": "Field indicating if the user is enrolled in marketing emails" }\n ]\n}")

implicit val config:EmbeddedKafkaConfig = EmbeddedKafkaConfig(
kafkaPort = 9092, schemaRegistryPort = 8081)

override def beforeAll(): Unit = {
super.beforeAll()
EmbeddedKafka.start()
}

override def afterAll(): Unit = {
EmbeddedKafka.stop()
super.afterAll()
}

"runs with embedded kafka and Schema Registry" should {

implicit val serializer: Serializer[GenericRecord] = {
  val props = Map(
    AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG -> "http://localhost:8081"
  )
  val ser = new KafkaAvroSerializer
  ser.configure(props.asJava, false)
  ser.asInstanceOf[Serializer[GenericRecord]]
}

implicit val deSerializer: Deserializer[GenericRecord] = {
  val props = Map(
    AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG -> "http://localhost:8081"
  )
  val ser = new KafkaAvroDeserializer()
  ser.configure(props.asJava, false)
  ser.asInstanceOf[Deserializer[GenericRecord]]
}
val topic ="input_topic"


"work" in {
  val record: GenericRecord = new GenericData.Record(schema)
  record.put("first_name", "yada")
  record.put("automated_email", false)
  EmbeddedKafka.withProducer[String,GenericRecord,Unit]( producer => {
    producer.send(new ProducerRecord[String,GenericRecord](topic,"abc",record))
    producer.flush()
  })

  val listOfTopics = Collections.singletonList(topic)
  EmbeddedKafka.withConsumer[String,GenericRecord,Unit](consumer => {
    consumer.subscribe(listOfTopics)
    var i = 10
    while(i>=0)
      {
        val records = consumer.poll(Duration.ofMillis(1000L))
        records.forEach(record => println("consumed message:" + record.key() +", " + record.value()))
        i-=1
      }
  })

}

}

}`

[francescopellegrini]

I just tried running your code and it works for me... here it is, reformatted and with the imports I used.

import io.confluent.kafka.serializers.{
  AbstractKafkaSchemaSerDeConfig,
  KafkaAvroDeserializer,
  KafkaAvroSerializer
}
import io.github.embeddedkafka.Codecs._
import org.apache.avro.Schema
import org.apache.avro.generic.{GenericData, GenericRecord}
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.serialization.{Deserializer, Serializer}
import org.scalatest.BeforeAndAfterAll
import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpecLike

import java.time.Duration
import java.util.Collections
import scala.jdk.CollectionConverters._

class AvroMockTesting extends AnyWordSpecLike with Matchers with BeforeAndAfterAll {

  val schema: Schema = new Schema.Parser().parse(
    """{ "type": "record", "namespace": "com.example", "name": "Customer", "version": "1", "fields": [ { "name": "first_name", "type": "string", "doc": "First Name of Customer" }, { "name": "automated_email", "type": "boolean", "default": true, "doc": "Field indicating if the user is enrolled in marketing emails" } ]}"""
  )

  implicit val config: EmbeddedKafkaConfig =
    EmbeddedKafkaConfig(kafkaPort = 9092, schemaRegistryPort = 8081)

  override def beforeAll(): Unit = {
    super.beforeAll()
    EmbeddedKafka.start()
  }

  override def afterAll(): Unit = {
    EmbeddedKafka.stop()
    super.afterAll()
  }

  "runs with embedded kafka and Schema Registry" should {

    implicit val serializer: Serializer[GenericRecord] = {
      val props = Map(
        AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG -> "http://localhost:8081"
      )
      val ser = new KafkaAvroSerializer
      ser.configure(props.asJava, false)
      ser.asInstanceOf[Serializer[GenericRecord]]
    }

    implicit val deSerializer: Deserializer[GenericRecord] = {
      val props = Map(
        AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG -> "http://localhost:8081"
      )
      val ser = new KafkaAvroDeserializer()
      ser.configure(props.asJava, false)
      ser.asInstanceOf[Deserializer[GenericRecord]]
    }
    val topic = "input_topic"

    "work" in {
      val record: GenericRecord = new GenericData.Record(schema)
      record.put("first_name", "yada")
      record.put("automated_email", false)
      EmbeddedKafka.withProducer[String, GenericRecord, Unit](producer => {
        producer.send(
          new ProducerRecord[String, GenericRecord](topic, "abc", record)
        )
        producer.flush()
      })

      val listOfTopics = Collections.singletonList(topic)
      EmbeddedKafka.withConsumer[String, GenericRecord, Unit](consumer => {
        consumer.subscribe(listOfTopics)
        var i = 10
        while (i >= 0) {
          val records = consumer.poll(Duration.ofMillis(1000L))
          records.forEach(record =>
            println("consumed message:" + record.key() + ", " + record.value())
          )
          i -= 1
        }
      })

    }
  }

}

[tharunrajguptayada]

import io.github.embeddedkafka.schemaregistry.EmbeddedKafkaConfig
import io.github.embeddedkafka.schemaregistry.EmbeddedKafka

I have tried with these imports , but still getting the same error, may be i will try with another windows machine and check.

Thanks for your response sir