package backtype.storm.contrib.cassandra.bolt;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.cassandra.service.CassandraHostConfigurator;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.Mutator;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import backtype.storm.contrib.cassandra.bolt.BatchingCassandraBolt.AckStrategy;
import backtype.storm.contrib.cassandra.bolt.determinable.ColumnFamilyDeterminable;
import backtype.storm.contrib.cassandra.bolt.determinable.DefaultColumnFamilyDeterminable;
import backtype.storm.contrib.cassandra.bolt.determinable.DefaultRowKeyDeterminable;
import backtype.storm.contrib.cassandra.bolt.determinable.RowKeyDeterminable;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;

@SuppressWarnings("serial")
public class DefaultBatchingCassandraBolt extends BatchingCassandraBolt implements
		CassandraConstants {
    private RowKeyDeterminable rkDeterminable;
	public DefaultBatchingCassandraBolt(
                    ColumnFamilyDeterminable cfDeterminable,
                    RowKeyDeterminable rkDeterminable) {
        super(cfDeterminable);
        this.rkDeterminable = rkDeterminable;
    }

    public DefaultBatchingCassandraBolt(String columnFamily, String rowKey) {
        this(new DefaultColumnFamilyDeterminable(columnFamily), new DefaultRowKeyDeterminable(rowKey));
    }

    private static final Logger LOG = LoggerFactory
			.getLogger(DefaultBatchingCassandraBolt.class);


    @Override
    public void executeBatch(List<Tuple> inputs) {
        ArrayList<Tuple> tuplesToAck = new ArrayList<Tuple>();
        try {
            Mutator<String> mutator = HFactory.createMutator(this.keyspace,
                    new StringSerializer());
            for (Tuple input : inputs) {
                String columnFamily = this.cfDeterminable
                        .determineColumnFamily(input);
                Object rowKey = this.rkDeterminable.determineRowKey(input);
                Fields fields = input.getFields();
                for (int i = 0; i < fields.size(); i++) {
                    mutator.addInsertion(rowKey.toString(), columnFamily,
                            HFactory.createStringColumn(fields.get(i), input
                                    .getValue(i).toString()));
                }
                tuplesToAck.add(input);
            }
            mutator.execute();

        } catch (Throwable e) {
            LOG.warn("Unable to write batch.", e);
        } finally {
            if (this.ackStrategy == AckStrategy.ACK_ON_WRITE) {
                for (Tuple tupleToAck : tuplesToAck) {
                    this.collector.ack(tupleToAck);
                }
            }
        }

    }


}

@@ -2754,16 +2754,219 @@
 ecute();
+%0A            if (this.ackStrategy == AckStrategy.ACK_ON_WRITE) %7B%0A                for (Tuple tupleToAck : tuplesToAck) %7B%0A                    this.collector.ack(tupleToAck);%0A                %7D%0A            %7D
 %0A%0A      
@@ -3046,28 +3046,8 @@
 e);%0A
-        %7D finally %7B%0A
     
@@ -3188,35 +3188,36 @@
  this.collector.
-ack
+fail
 (tupleToAck);%0A  

move ack loop out of finally block so tuples are acked only when cassandra write is successful add fail loop to catch block to tuples are failed quickly when writes fail.