@SuppressWarnings({ "unchecked", "rawtypes" })
    @Override
    public State makeState(Map conf, IMetricsContext metrics, int partitionIndex, int numPartitions) {
        LOG.info("Making new CassandraState object for cluster " + cassandraClusterId + ": partition [" + partitionIndex + "] of [" + numPartitions + "]");
        AstyanaxClient client = AstyanaxClientFactory.getInstance(cassandraClusterId, (Map)conf.get(cassandraClusterId));
        int batchMaxSize = Utils.getInt(Utils.get(conf, StormCassandraConstants.CASSANDRA_BATCH_MAX_SIZE,
                CassandraState.DEFAULT_MAX_BATCH_SIZE));
        return new CassandraState(client, batchMaxSize, this.exceptionHandler);
    }
@ SuppressWarnings ( { "unchecked" , "rawtypes" } ) @ Override public State makeState ( Map conf , IMetricsContext metrics , int partitionIndex , int numPartitions ) { LOG . info ( "Making new CassandraState object for cluster " + cassandraClusterId + ": partition [" + partitionIndex + "] of [" + numPartitions + "]" ) ; AstyanaxClient client = AstyanaxClientFactory . getInstance ( cassandraClusterId , ( Map ) conf . get ( cassandraClusterId ) ) ; int batchMaxSize = Utils . getInt ( Utils . get ( conf , StormCassandraConstants . CASSANDRA_BATCH_MAX_SIZE , CassandraState . DEFAULT_MAX_BATCH_SIZE ) ) ; return new CassandraState ( client , batchMaxSize , this . exceptionHandler ) ; }
/* (non-Javadoc)
@see storm.trident.state.StateFactory#makeState(java.util.Map, backtype.storm.task.IMetricsContext, int, int)