Composing a custom producer for Kafka
I downloaded Kafka from https://www.apache.org/dyn/closer.cgi?path=/kafka/0.8.1.1/kafka_2.8.0-0.8.1.1.tgz and configured a kafka cluster on my machine with virtual machines. The cluster is working fine - it has been tested using a console producer and consumer with the kafka package.
Now I have applied my own Producer class for Kafka. But I couldn't figure out how to compile this class and what the dependencies are.
Questions
-
Can anyone explain how I need to collect dependencies for Producer by creating a class and running it?
-
Do I need sbt to create it? I couldn't find a single online resource that clearly explained how to go about creating a custom kafka producer class.
The following are the packages imported in the Producer class:
org.apache.kafka.clients.producer.Callback;
org.apache.kafka.clients.producer.KafkaProducer;
org.apache.kafka.clients.producer.ProducerConfig;
org.apache.kafka.clients.producer.ProducerRecord;
org.apache.kafka.clients.producer.RecordMetadata;
org.apache.kafka.common.record.Records
Thank you in advance
source to share
I developed my own Kafka producer, like the Maven project I was using:
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>0.8.2.0</version>
</dependency>
The import that I used:
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringSerializer;
Snippet of my manufacturer's dispatch code:
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, zkConnection);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
byte[] byteData = null;
File myInputFile = new File(...);
try (InputStream inputStream = new FileInputStream(myInputFile)) {
byteData = IOUtils.toByteArray(inputStream);
}
try (KafkaProducer<String, byte[]> producer = new KafkaProducer<String, byte[]>(props)) {
producer.send(new ProducerRecord<String, byte[]>(topic, byteData));
}
source to share