/*
 * Copyright 2010 LinkedIn
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package kafka.examples;

import kafka.javaapi.message.ByteBufferMessageSet;
import kafka.javaapi.producer.SyncProducer;
import kafka.message.Message;
import kafka.producer.SyncProducerConfig;
import java.util.Properties;
import java.util.ArrayList;
import java.util.List;

public class Producer extends Thread
{
  private final SyncProducer producer;
  private final String topic;
  private final Properties props = new Properties();

  public Producer(String topic)
  {
    props.put("host", KafkaProperties.kafkaServerURL);
    props.put("port", String.valueOf(KafkaProperties.kafkaServerPort));
    props.put("buffer.size", String.valueOf(KafkaProperties.kafkaProducerBufferSize));
    props.put("connect.timeout.ms", String.valueOf(KafkaProperties.connectionTimeOut));
    props.put("reconnect.interval", String.valueOf(KafkaProperties.reconnectInterval));
    producer = new SyncProducer(new SyncProducerConfig(props));
    this.topic = topic; 
    
  }
  
  public void run() {
    int messageNo = 1;
    while(true)
    {
      String messageStr = new String("Message_" + messageNo);
      Message message = new Message(messageStr.getBytes());
      List<Message> messageList = new ArrayList<Message>();
      messageList.add(message);
      ByteBufferMessageSet set = new ByteBufferMessageSet(messageList);
      producer.send(topic, set);
      messageNo++;
    }
  }

}

@@ -615,59 +615,8 @@
 s;%0A%0A
-import kafka.javaapi.message.ByteBufferMessageSet;%0A
 impo
@@ -641,28 +641,24 @@
 roducer.
-Sync
 Producer
 ;%0Aimport
@@ -653,38 +653,12 @@
 ucer
-;%0Aimport kafka.message.Message
+Data
 ;%0Aim
@@ -677,20 +677,16 @@
 roducer.
-Sync
 Producer
@@ -725,59 +725,8 @@
 ies;
-%0Aimport java.util.ArrayList;%0Aimport java.util.List;
 %0A%0Apu
@@ -778,28 +778,64 @@
 e final 
-SyncProducer
+kafka.javaapi.producer.Producer%3CInteger, String%3E
  produce
@@ -976,423 +976,299 @@
 ut(%22
-host%22, K
+serializer.class%22, %22k
 afka
-Properties.kafkaServerURL);%0A    props.put(%22port%22, String.valueOf(KafkaProperties.kafkaServerPort));%0A    props.put(%22buffer.size%22, String.valueOf(KafkaProperties.kafkaProducerBufferSize));%0A    props.put(%22connect.timeout.ms%22, String.valueOf(KafkaProperties.connectionTimeOut));%0A    props.put(%22reconnect.interval%22, String.valueOf(KafkaProperties.reconnectInterval));%0A    producer = new SyncProducer
+.serializer.StringEncoder%22);%0A    props.put(%22zk.connect%22, %22localhost:2181%22);%0A    // Use random partitioner. Don't need the key type. Just set it to Integer.%0A    // The message is of type String.%0A    producer = new kafka.javaapi.producer.Producer%3CInteger, String%3E
 (new 
-Sync
 Prod
@@ -1314,14 +1314,8 @@
 pic;
- %0A    
 %0A  %7D
@@ -1457,256 +1457,74 @@
     
-Message message = new Message(messageStr.getBytes());%0A      List%3CMessage%3E messageList = new ArrayList%3CMessage%3E();%0A      messageList.add(message);%0A      ByteBufferMessageSet set = new ByteBufferMessageSet(messageList);%0A      producer.send(topic, set
+producer.send(new ProducerData%3CInteger, String%3E(topic, messageStr)
 );%0A 

change java example to use the new producer api