import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.time.Instant;
import java.util.Properties;

public class KafkaProducerApp {

  public static void main(String[] args) {
    Properties props = new Properties();
    KafkaProducerCommon.configure(props);

    KafkaProducer<String, String> myProducer = new KafkaProducer<String, String>(props);

    try {
      for (int i = 0; i < 100; i++) {
        Instant ts = Instant.now();
        Double ss = ts.toEpochMilli() + ts.getNano() / 1E9;
        ProducerRecord myMessage = new ProducerRecord(KafkaCommon.TOPIC, String.format("%3d : %09.3f", i, ss));
        myProducer.send(myMessage); // Best practice: wrap in try..catch.
      }
    } catch (Exception e) {
      e.printStackTrace();
    } finally {
      myProducer.close();
    }
  }

  public static void createMessagesExample() {
    // Message with only the required properties.
    ProducerRecord myMessage = new ProducerRecord(
      KafkaCommon.TOPIC, // Topic to which the record will be sent.
      "My Message 1" // Message content, matching the serializer type for value
    );

    // Non-matching type: runtime exception
    // ProducerRecord myBadMessage = new ProducerRecord(TOPIC, 3.14159);

    ProducerRecord myPartitionedMessage = new ProducerRecord(
      KafkaCommon.TOPIC, // String Topic
      1,  // Integer Partition
      "My Message 1" // String Message
    );

    ProducerRecord myKeyedMessage = new ProducerRecord(
      KafkaCommon.TOPIC, // String Topic
      "Course-001",  // K key
      "My Message 1" // String Message
    );

    // Adding optional properties
    ProducerRecord msg3 = new ProducerRecord(
      KafkaCommon.TOPIC, // String Topic
      1, // Integer Partition
      124535353325L, // Long timestamp, added in Kafka 0.10.
      "Course-001", // K key: basis to determine partitioning strategy. Don't use blank or NULL.
      // Key may contain additional message information, but adds overhead, depends on serializer.
      "My Message 1" // V value
    );
    // The actual TS being send with the message is defined in server.properties:
    // log.message.timestamp.type = [CreateTime, LogAppendTime]
    // - CreateTime: producer-set timestamp is used
    // - LogAppendtime: broker-set to the time when the message is appended to the commit log. Overrides the producet-set one.
  }
}