Posts

Showing posts with the label apache-kafka

Kafka : Generating unique IDs for strings across partitions

Kafka : Generating unique IDs for strings across partitions I'm trying to asses if Kafka could be used to scale-out our current solution. I can identify partitions easily. Currently, the requirement is there to be 1500 partitions, each having 1-2 events per second, but future might go as high as 10000 partitions. But there is one part of our solution which I don't know how would be solved in Kafka. The problem is that each message contains a string and I want to assign unique ID to each string across whole topic. So same strings have same ID while different strings have different IDs. The IDs don't need to be sequential, nor do they need to be always-growing. The IDs will then be used down-stream as unique keys to identify those strings. The strings can be hundreds of characters long, so I don't think they would make efficient keys. More advanced usage would be where messages might have different "kinds" of strings, so there would be multiple unique sequences ...

Getting Bootstrap broker ip:9092 disconnected error from kafka spout

Getting Bootstrap broker ip:9092 disconnected error from kafka spout Versions: "org.apache.storm" % "storm-kafka-client" % "1.2.1" "org.apache.storm" % "storm-core" % "1.2.1" % "compile" Kafka: 0.10.1.0 I am getting following error/warnings, running in localCluster, from my kafka spout: 2018-06-28 00:00:34,930 AppInfoParser [INFO] Kafka version : 0.10.1.0 2018-06-28 00:00:34,930 AppInfoParser [INFO] Kafka commitId : 3402a74efb23d1d4 2018-06-28 00:00:34,931 WARN NetworkClient [Thread-40-KafkaSpout-executor[12 12]] Bootstrap broker ip1:9092 disconnected 2018-06-28 00:00:35,092 WARN NetworkClient [Thread-40-KafkaSpout-executor[12 12]] Bootstrap broker ip2:9092 disconnected 2018-06-28 00:00:35,251 WARN NetworkClient [Thread-40-KafkaSpout-executor[12 12]] Bootstrap broker ip3:9092 disconnected 2018-06-28 00:00:35,524 WARN NetworkClient [Thread-40-KafkaSpout-executor[12 12]] Bootstrap broker ip4:9092 disconnected 2018-06-2...

Trouble with deserializing Avro data in Scala

Trouble with deserializing Avro data in Scala I am building an Apache Flink application in Scala which reads streaming data from a Kafka bus and then performs summarizing operations on it. The data from Kafka is in Avro format and needs a special Deserialization class. I found this scala class AvroDeserializationScehema (http://codegists.com/snippet/scala/avrodeserializationschemascala_saveveltri_scala): package org.myorg.quickstart import org.apache.avro.io.BinaryDecoder import org.apache.avro.io.DatumReader import org.apache.avro.io.DecoderFactory import org.apache.avro.reflect.ReflectDatumReader import org.apache.avro.specific.{SpecificDatumReader, SpecificRecordBase} import org.apache.flink.api.common.typeinfo.TypeInformation import org.apache.flink.api.java.typeutils.TypeExtractor import org.apache.flink.api.common.serialization._ import java.io.IOException class AvroDeserializationSchema[T](val avroType: Class[T]) extends DeserializationSchema[T] { private var reader: DatumRead...

How to check which schema registry instances are connected to a Zookeeper quorum

How to check which schema registry instances are connected to a Zookeeper quorum I am running two instances of schema registry that point to the same zookeeper quorum Here are the properties of the first instance: port=8081 # The address the socket server listens on. # FORMAT: # listeners = listener_name://host_name:port # EXAMPLE: # listeners = PLAINTEXT://your.host.name:9092 listeners=http://0.0.0.0:8081 # Zookeeper connection string for the Zookeeper cluster used by your Kafka cluster # (see zookeeper docs for details). # This is a comma separated host:port pairs, each corresponding to a zk # server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002". kafkastore.connection.url=host-1:2181,host-2:2181,host-3:2181 # Alternatively, Schema Registry can now operate without Zookeeper, handling all coordination via # Kafka brokers. Use this setting to specify the bootstrap servers for your Kafka cluster and it # will be used both for selecting the master schema regis...

How does Kafka treat negative number in partition in ProducerRecord in Kafka Client 0.10?

How does Kafka treat negative number in partition in ProducerRecord in Kafka Client 0.10? In Kafka Client 1.1.0 a check is placed to not allow negative as partition in ProducerRecord . partition ProducerRecord https://github.com/apache/kafka/blob/1.1/clients/src/main/java/org/apache/kafka/clients/producer/ProducerRecord.java#L73 But I don't see any such check in ProducerRecord in kafka client 0.10.0. ProducerRecord https://github.com/apache/kafka/blob/0.10.0/clients/src/main/java/org/apache/kafka/clients/producer/ProducerRecord.java#L62 Can someone let me know how does Kafka handle the negative number in partition in ProducerRecord in 0.10? Will the partition be converted to positive number ? If yes can some one point me to the code / logic of the conversion? partition ProducerRecord 1 Answer 1 With the 0.10.0 Kafka Producer, a negative number for the partition is also rejected. The check wa...