code: @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);
      }
docs: /* (non-Javadoc)
  @see storm.trident.state.StateFactory#makeState(java.util.Map, backtype.storm.task.IMetricsContext, int, int)