anchor.diff:  
   image::2016-10-10-Kafka-Java-Client/kafka_producer.svg[Producer]
   
  +When no key is provided, the producer uses a random partition.
  +In short, the producer is able to load balance writes to all partitions, and associated nodes.
  +
   This producer is initialized and configured with some properties.
   
   [source,java]
   ----
   Map<String, Object> producerConfig = new HashMap<>();
   producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka1:9092,kafka2:9092"); <1>
  -producerConfig.put(ProducerConfig.ACKS_CONFIG, "1"); <4>
  -producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); <5>
  +producerConfig.put(ProducerConfig.ACKS_CONFIG, "1"); <5>
  +producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); <4>
   producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
   try (Producer<String, String> producer = new KafkaProducer<>(producerConfig)) {
       ProducerRecord<String, String> record = new ProducerRecord<>("the_topic", "the_key","The Message"); <2>
       Future<RecordMetadata> futureMetadata = producer.send(record);<3>
  -    RecordMetadata metadata = producer.get(); <4>
  +    RecordMetadata metadata = producer.get(); <5>
   }
   ----
   <1> Connect to these Kafka connect nodes.
  
positive.code: }
  ----
  
  This offset can even be moved forward (to skip records) and backwards (to replay records):
  
  [source,java]
  ----
      consumer.seekToBeginning(consumer.assignment());
  ----
  
  == Using the Spring framework
  
  You may have noticed that the consumer API is a pull API.
  In a real application you'll have to create a consuming loop in separate thread,
  
negative.code: }
  ----
  
  This offset commit can also be manual in order to ensure messages are acknowledged once they have been processed.
  
  [source,java]
  ----
  