/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 storm.kafka.trident;

import backtype.storm.task.OutputCollector;
import backtype.storm.topology.FailedException;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import org.apache.commons.lang.Validate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import storm.kafka.trident.mapper.TridentTupleToKafkaMapper;
import storm.kafka.trident.selector.KafkaTopicSelector;
import storm.trident.operation.TridentCollector;
import storm.trident.state.State;
import storm.trident.tuple.TridentTuple;

import java.util.List;
import java.util.Map;
import java.util.Properties;

public class TridentKafkaState implements State {
    private static final Logger LOG = LoggerFactory.getLogger(TridentKafkaState.class);

    public static final String KAFKA_BROKER_PROPERTIES = "kafka.broker.properties";

    private Producer producer;
    private OutputCollector collector;

    private TridentTupleToKafkaMapper mapper;
    private KafkaTopicSelector topicSelector;

    public TridentKafkaState withTridentTupleToKafkaMapper(TridentTupleToKafkaMapper mapper) {
        this.mapper = mapper;
        return this;
    }

    public TridentKafkaState withKafkaTopicSelector(KafkaTopicSelector selector) {
        this.topicSelector = selector;
        return this;
    }

    @Override
    public void beginCommit(Long txid) {
        LOG.debug("beginCommit is Noop.");
    }

    @Override
    public void commit(Long txid) {
        LOG.debug("commit is Noop.");
    }

    public void prepare(Map stormConf) {
        Validate.notNull(mapper, "mapper can not be null");
        Validate.notNull(topicSelector, "topicSelector can not be null");
        Map configMap = (Map) stormConf.get(KAFKA_BROKER_PROPERTIES);
        Properties properties = new Properties();
        properties.putAll(configMap);
        ProducerConfig config = new ProducerConfig(properties);
        producer = new Producer(config);
    }

    public void updateState(List<TridentTuple> tuples, TridentCollector collector) {
        String topic = null;
        for (TridentTuple tuple : tuples) {
            try {
                topic = topicSelector.getTopic(tuple);

                if(topic != null) {
                    producer.send(new KeyedMessage(topic, mapper.getKeyFromTuple(tuple),
                            mapper.getMessageFromTuple(tuple)));
                } else {
                    LOG.warn("skipping key = " + mapper.getKeyFromTuple(tuple) + ", topic selector returned null.");
                }
            } catch (Exception ex) {
                String errorMsg = "Could not send message with key = " + mapper.getKeyFromTuple(tuple)
                        + " to topic = " + topic;
                LOG.warn(errorMsg, ex);
                throw new FailedException(errorMsg, ex);
            }
        }
    }
}

@@ -929,89 +929,180 @@
 ort 
-kafka.javaapi.producer.Producer;%0Aimport kafka.producer.KeyedMessage;%0Aimport kafka
+org.apache.commons.lang.Validate;%0Aimport org.apache.kafka.clients.producer.Callback;%0Aimport org.apache.kafka.clients.producer.KafkaProducer;%0Aimport org.apache.kafka.clients
 .pro
@@ -1115,22 +1115,22 @@
 Producer
-Config
+Record
 ;%0Aimport
@@ -1141,37 +1141,53 @@
 .apache.
-commons.lang.Validate
+kafka.clients.producer.RecordMetadata
 ;%0Aimport
@@ -1790,16 +1790,21 @@
 private 
+Kafka
 Producer
@@ -2795,44 +2795,36 @@
     
-P
+p
 roducer
-Config config
  = new 
+Kafka
 Producer
 Conf
@@ -2823,14 +2823,8 @@
 ucer
-Config
 (pro
@@ -2837,49 +2837,8 @@
 s);%0A
-        producer = new Producer(config);%0A
     
@@ -2937,42 +2937,19 @@
     
-String topic = null;%0A        for (
+for (final 
 Trid
@@ -2991,29 +2991,20 @@
     
-try %7B%0A               
+final String
  top
@@ -3039,19 +3039,16 @@
 tuple);%0A
-%0A  
         
@@ -3051,18 +3051,16 @@
         
-  
 if(topic
@@ -3083,28 +3083,24 @@
             
-    
 producer.sen
@@ -3109,20 +3109,22 @@
 new 
-KeyedMessage
+ProducerRecord
 (top
@@ -3182,20 +3182,16 @@
         
-    
 mapper.g
@@ -3220,152 +3220,157 @@
 le))
-);%0A                %7D else %7B%0A                    LOG.warn(%22skipping key = %22 + mapper.getKeyFromTuple(tuple) + %22, topic selector returned null.%22);
+,new Callback() %7B%0A                            @Override%0A                            public void onCompletion(RecordMetadata metadata, Exception ex) %7B
 %0A   
@@ -3374,34 +3374,32 @@
                 
-%7D%0A
             %7D ca
@@ -3398,33 +3398,48 @@
     
-%7D catch (Exception ex) %7B%0A
+    if(ex != null)%7B%0A                    
     
@@ -3508,91 +3508,123 @@
  = %22
- + mapper.getKeyFromTuple(tuple)%0A                        + %22 to topic = %22 + topic;%0A
+%0A                                    + mapper.getKeyFromTuple(tuple) + %22 to topic = %22 + topic;%0A                    
     
@@ -3675,16 +3675,36 @@
         
+                    
 throw ne
@@ -3728,32 +3728,258 @@
 (errorMsg, ex);%0A
+                                %7D%0A                            %7D%0A                        %7D);%0A            %7D else %7B%0A                LOG.warn(%22skipping key = %22 + mapper.getKeyFromTuple(tuple) + %22, topic selector returned null.%22);%0A
             %7D%0A  

use new kafka producer api