blog

  // psvm short hand for main method
  public static void main(String[] args) {

      // create producer propertis
      Properties properties = new Properties();
      properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
      properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
      properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

      // create the producer
      KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);

      // create a producer record
      ProducerRecord<String, String> record = new ProducerRecord<String, String>("first_topic", "hello world");

      // send data - asynchronous
      producer.send(record);
      // flush data
      producer.flush();
      // flush and close producer
      producer.close();
  }

  public static void main(String[] args) {
    final Logger logger = LoggerFactory.getLogger(ProducerDemo.class);

    // create producer propertis
    Properties properties = new Properties();
    properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
    properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

    // create the producer
    KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);

    // create a producer record
    ProducerRecord<String, String> record = new ProducerRecord<String, String>("first_topic", "hello world");


    // send data - asynchronous
    producer.send(record, new Callback() {
        // execute every time a record is successfully sent or an exception is thrown
        public void onCompletion(RecordMetadata recordMetadata, Exception e) {
            if (e == null) {
                // the record was successfully sent
                logger.info("Received new metadata. \n" +
                            "Topic: " + recordMetadata.topic() + "\n" +
                            "Partition: " + recordMetadata.partition() + "\n" +
                            "Offset: " + recordMetadata.offset() + "\n" +
                            "Timestamp: " + recordMetadata.timestamp());
            } else {
                logger.error("Error while producing", e);
            }
        }
    });

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

  public static void main(String[] args) throws ExecutionException, InterruptedException {
    final Logger logger = LoggerFactory.getLogger(ProducerDemo.class);

    // create producer propertis
    Properties properties = new Properties();
    properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
    properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

    // create the producer
    KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);

    for (int i=0; i<10; i++) {
        String topic = "first_topic";
        String value  = "hello world " + i;
        String key = "id_" + i;

        // create a producer record
        ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, key, value);

        logger.info("Key: " + key);

        // send data - asynchronous
        producer.send(record, new Callback() {
            // execute every time a record is successfully sent or an exception is thrown
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                if (e == null) {
                    // the record was successfully sent
                    logger.info("Received new metadata. \n" +
                                "Topic: " + recordMetadata.topic() + "\n" +
                                "Partition: " + recordMetadata.partition() + "\n" +
                                "Offset: " + recordMetadata.offset() + "\n" +
                                "Timestamp: " + recordMetadata.timestamp());
                } else {
                    logger.error("Error while producing", e);
                }
            }
        });
    }

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

  public static void main(String[] args) {

    Logger logger = LoggerFactory.getLogger(ConsumerDemo.class.getName());

    // create consumer configs
    Properties properties = new Properties();
    properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
    properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "my-first-application");
    // can also be latest -> read only new message, none -> throw an error if there is no message has been saved
    properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

    // create consumer
    KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);

    // subscribe consumer to out topics
    // .singleton means only subscribe to one topic. If multiple, we can do Arrays.asList("topic1", "topic2" ...)
    consumer.subscribe(Collections.singleton("first_topic"));

    // poll for new data
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

        for (ConsumerRecord<String, String> record : records) {
            logger.info("Key: "+ record.key() + ", value: " + record.value());
            logger.info("Partition: " + record.partition() + ", Offset: " + record.offset());
        }
    }
}

public static void main(String[] args) {
    new ConsumerDemo().run();
}

private ConsumerDemo() {

}

private void run() {
    Logger logger = LoggerFactory.getLogger(ConsumerDemo.class.getName());

    String bootstrapServers = "127.0.0.1:9092";
    String groupId = "my-first-application";
    String topic = "first_topic";
    CountDownLatch latch = new CountDownLatch(1);

    Runnable myConsumerThread = new ConsumerThread(bootstrapServers, groupId, topic, latch);

    Thread myThread = new Thread(myConsumerThread);
    myThread.start();

    Runtime.getRuntime().addShutdownHook(new Thread(()-> {
        logger.info("Caught shuntdown hook");
        ((ConsumerThread)myConsumerThread).shutdown();
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }));

    try {
        latch.await();
    } catch (InterruptedException e) {
        logger.error("Application got interrupted", e);
    } finally {
        logger.info("Application is closing");
    }
}


public class ConsumerThread implements Runnable {

    private CountDownLatch latch;
    private KafkaConsumer<String, String> consumer;
    private Logger logger = LoggerFactory.getLogger(ConsumerThread.class.getName());

    public ConsumerThread(String bootstrapServers, String groupId, String topic, CountDownLatch latch) {
        this.latch = latch;

        // create consumer configs
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "my-first-application");
        // can also be latest -> read only new message, none -> throw an error if there is no message has been saved
        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        consumer = new KafkaConsumer<String, String>(properties);
        // subscribe consumer to out topics
        // .singleton means only subscribe to one topic. If multiple, we can do Arrays.asList("topic1", "topic2" ...)
        consumer.subscribe(Collections.singleton("first_topic"));
    }

    @Override
    public void run() {
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

                for (ConsumerRecord<String, String> record : records) {
                    logger.info("Key: "+ record.key() + ", value: " + record.value());
                    logger.info("Partition: " + record.partition() + ", Offset: " + record.offset());
                }
            }
        } catch (WakeupException e) {
            logger.info("Received shutdown signal!");
        } finally {
            consumer.close();
            latch.countDown();
        }

    }

    public void shutdown() {
        // a special method to interrupt consumer.poll()
        // it will throw the wakeupexception
        consumer.wakeup();
    }
}

 public static void main(String[] args) {

    Logger logger = LoggerFactory.getLogger(ConsumerDemo.class.getName());

    // create consumer configs
    Properties properties = new Properties();
    properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
    properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    // can also be latest -> read only new message, none -> throw an error if there is no message has been saved
    properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

    // create consumer
    KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);

    // assign and seek are mostly used to replay data or fetch a specific message

    // assign
    TopicPartition partitionToReadFrom = new TopicPartition("first_topic", 0);
    long offsetToReadFrom = 5L;
    consumer.assign(Arrays.asList(partitionToReadFrom));

    // seek
    consumer.seek(partitionToReadFrom, offsetToReadFrom);

    int numberOfMessagesToRead = 10;
    boolean keepOnReading = true;
    int numberOfMessagesReadSoFar = 0;

    // poll for new data
    while (keepOnReading) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

        for (ConsumerRecord<String, String> record : records) {
            numberOfMessagesReadSoFar += 1;
            logger.info("Key: "+ record.key() + ", value: " + record.value());
            logger.info("Partition: " + record.partition() + ", Offset: " + record.offset());
            if (numberOfMessagesReadSoFar >= numberOfMessagesToRead) {
                keepOnReading = false;
                break;
            }
        }
    }

    logger.info("Exit application");
}