import me.prettyprint.cassandra.serializers.BytesArraySerializer; //导入依赖的package包/类
private HColumn<String,byte[]> querySingleColumnWithRetry(final ShardKey shard,final String actorId) {
    // try three times, and log a warning when we exceed the readExecutionThreshold
    final long startTime = currentTimeMillis();
    int attemptsRemaining = 3;
    try {
        while (true) {
            attemptsRemaining--;
            try {
                return columnFamilyTemplate.querySingleColumn(createKey(shard), actorId, BytesArraySerializer.get());
            } catch (HTimedOutException | HPoolRecoverableException e) {
                if (attemptsRemaining <= 0) {
                    throw e;
                }
            }
        }
    } finally {
        final long endTime = currentTimeMillis();
        if((endTime - startTime) > readExecutionThresholdMillis) {
            logger.warn(format("Cassandra read operation took %d msecs (%d retries) for actorId [%s] on shard [%s]",(endTime - startTime),(2 - attemptsRemaining),actorId,shard.toString()));
        }
    }
}
 

import me.prettyprint.cassandra.serializers.BytesArraySerializer; //导入依赖的package包/类
private <K> RowSerializer<K, String> getRowSerializerForEntity(EntityMetadata metadata) {
    Map<String, Serializer> columnSerializers = new HashMap<String, Serializer>();

    if (metadata.getEntitySerializer() != null) {
        AbstractHerculesSerializer entitySerializer = serializerProvider.getSerializer(metadata.getEntitySerializer(), metadata.getEntityClass());
        columnSerializers.put(SERIALIZED_ENTITY_TOP_KEY, entitySerializer);
    } else {
        for (Map.Entry<String, Class> entry : metadata.getColumnClasses().entrySet()) {
            Serializer fieldSerializer = getFieldSerializer(entry.getKey(), metadata);
            columnSerializers.put(entry.getKey(), fieldSerializer);
        }
    }

    columnSerializers.put(METADATA_COLUMN_NAME, BytesArraySerializer.get());

    return new ColumnRowSerializer<K, String>(this.<K>getPrimaryKeySerializer(metadata), StringSerializer.get(), columnSerializers);
}
 

import me.prettyprint.cassandra.serializers.BytesArraySerializer; //导入依赖的package包/类
@Override
public List<T> fromByteBuffer(ByteBuffer byteBuffer) {
  if (byteBuffer == null) {
    return null;
  }
  ArrayList<T> array = new ArrayList<>();
  while (true) {
    T element = null;
    try {
      if (size > 0) {
        element = elementSerializer.fromByteBuffer(byteBuffer);
      }
      else {
        int n = IntegerSerializer.get().fromByteBuffer(byteBuffer);
        byte[] bytes = new byte[n];
        byteBuffer.get(bytes, 0, n);
        element = elementSerializer.fromByteBuffer( BytesArraySerializer.get().toByteBuffer(bytes) );
      }
    } catch (BufferUnderflowException e) {
      break;
    }
    if (element == null) {
      break;
    }
    array.add(element);
  }
  return array;
}
 

import me.prettyprint.cassandra.serializers.BytesArraySerializer; //导入依赖的package包/类
@Override
public ByteBuffer toByteBuffer(SpecificFixed fixed) {
  if (fixed == null) {
    return null;
  }
  byte[] bytes = fixed.bytes();
  if (bytes.length < 1) {
    return null;
  }
  return BytesArraySerializer.get().toByteBuffer(bytes);
}
 

import me.prettyprint.cassandra.serializers.BytesArraySerializer; //导入依赖的package包/类
@Override
public Map<CharSequence, T> fromByteBuffer(ByteBuffer byteBuffer) {
  if (byteBuffer == null) {
    return null;
  }
  Map<CharSequence, T> map = new HashMap<>();
  while (true) {
    CharSequence key = null;
    T value = null;
    try {
      int n = IntegerSerializer.get().fromByteBuffer(byteBuffer);
      byte[] bytes = new byte[n];
      byteBuffer.get(bytes, 0, n);
      key = CharSequenceSerializer.get().fromByteBuffer( BytesArraySerializer.get().toByteBuffer(bytes) );

      if (size > 0) {
        value = valueSerializer.fromByteBuffer(byteBuffer);
      }
      else {
        n = IntegerSerializer.get().fromByteBuffer(byteBuffer);
        bytes = new byte[n];
        byteBuffer.get(bytes, 0, n);
        value = valueSerializer.fromByteBuffer( BytesArraySerializer.get().toByteBuffer(bytes) );
      }
    } catch (BufferUnderflowException e) {
      break;
    }
    if (key == null) {
      break;
    }
    if (value == null) {
      break;
    }
    map.put(key, value);
  }
  return map;
}
 

import me.prettyprint.cassandra.serializers.BytesArraySerializer; //导入依赖的package包/类
@SuppressWarnings({ "rawtypes", "unchecked" })
public static <T> Serializer<T> getSerializer(Schema schema) {
  Serializer serializer = null;
  Type type = schema.getType();
  if (type.equals(Type.STRING)) {
    serializer = CharSequenceSerializer.get();
  } else if (type.equals(Type.BOOLEAN)) {
    serializer = BooleanSerializer.get();
  } else if (type.equals(Type.BYTES)) {
    serializer = ByteBufferSerializer.get();
  } else if (type.equals(Type.DOUBLE)) {
    serializer = DoubleSerializer.get();
  } else if (type.equals(Type.FLOAT)) {
    serializer = FloatSerializer.get();
  } else if (type.equals(Type.INT)) {
    serializer = IntegerSerializer.get();
  } else if (type.equals(Type.LONG)) {
    serializer = LongSerializer.get();
  } else if (type.equals(Type.FIXED)) {
    Class clazz = TypeUtils.getClass(schema);
    serializer = SpecificFixedSerializer.get(clazz);
    // serializer = SpecificFixedSerializer.get(schema);
  } else if (type.equals(Type.ARRAY)) {
    serializer = ListSerializer.get(schema.getElementType());
  } else if (type.equals(Type.MAP)) {
  	serializer = MapSerializer.get(schema.getValueType());
  } else if (type.equals(Type.UNION)){
    serializer = ByteBufferSerializer.get();
  } else if (type.equals(Type.RECORD)){
    serializer = BytesArraySerializer.get();
  } else {
    serializer = null;
  }
  return serializer;
}
 

import me.prettyprint.cassandra.serializers.BytesArraySerializer; //导入依赖的package包/类
@Override
public void createRequiredSchemaEntities() throws DataAccessLayerException {
	final String keyspace_name = _keyspace.getKeyspaceName();
	final KeyspaceDefinition ksdef = _factory.getCluster().describeKeyspace(keyspace_name);

	if (!hasColumnFamily(ksdef, _cf_name)) {

		final ColumnFamilyDefinition cfDef = HFactory.createColumnFamilyDefinition(keyspace_name, _cf_name);
		final Map<String, String> compressionOptions = new HashMap<String, String>();
		compressionOptions.put("sstable_compression", "SnappyCompressor");
		cfDef.setCompressionOptions(compressionOptions);
		
		cfDef.setColumnType(ColumnType.STANDARD);
		cfDef.setKeyValidationClass(ComparatorType.BYTESTYPE.getClassName());
		cfDef.setDefaultValidationClass(ComparatorType.BYTESTYPE.getClassName());
		cfDef.setCompactionStrategy("LeveledCompactionStrategy");

		if (_isBidirectional) {

			BasicColumnDefinition colDef = new BasicColumnDefinition();
			colDef.setName(BytesArraySerializer.get().toByteBuffer(COLUMN_NAME));
			colDef.setValidationClass(ComparatorType.BYTESTYPE.getClassName());
			colDef.setIndexType(ColumnIndexType.KEYS);
			colDef.setIndexName(_cf_name + "_val_idx");

			cfDef.addColumnDefinition(colDef);
		}

		_factory.getCluster().addColumnFamily(new ThriftCfDef(cfDef), true);
	}
}
 

import me.prettyprint.cassandra.serializers.BytesArraySerializer; //导入依赖的package包/类
/** Constructor. */
public CassandraStorageSystem(int port) throws StorageException {
	try {
		Cluster cluster = createCluster(port);
		if (cluster.describeKeyspace(KEYSPACE_NAME) == null) {
			createKeyspace(cluster);
		}
		keyspace = HFactory.createKeyspace(KEYSPACE_NAME, cluster);
		template = new ThriftColumnFamilyTemplate<byte[], String>(keyspace,
				COLUMN_FAMILY_NAME, BytesArraySerializer.get(),
				StringSerializer.get());
	} catch (HectorException e) {
		throw new StorageException(e);
	}
}
 

import me.prettyprint.cassandra.serializers.BytesArraySerializer; //导入依赖的package包/类
/** Implements a batch get operation for a list of keys. */
private List<byte[]> batchGetSubList(List<byte[]> keys) {
	MultigetSliceQuery<byte[], String, byte[]> multigetSliceQuery = HFactory
			.createMultigetSliceQuery(keyspace, BytesArraySerializer.get(),
					StringSerializer.get(), BytesArraySerializer.get());
	multigetSliceQuery
			.setColumnFamily(CassandraStorageSystem.COLUMN_FAMILY_NAME);
	multigetSliceQuery.setKeys(keys);
	multigetSliceQuery.setColumnNames(CassandraStorageSystem.COLUMN_NAME);

	Map<ByteArrayWrapper, Integer> keyToIndex = new HashMap<ByteArrayWrapper, Integer>();
	int index = 0;
	for (byte[] key : keys) {
		keyToIndex.put(new ByteArrayWrapper(key), index);
		index += 1;
	}

	List<byte[]> resultList = new ArrayList<byte[]>(
			Collections.<byte[]> nCopies(keys.size(), null));
	for (Row<byte[], String, byte[]> row : multigetSliceQuery.execute()
			.get()) {
		List<HColumn<String, byte[]>> columns = row.getColumnSlice()
				.getColumns();
		if (!columns.isEmpty()) {
			int i = keyToIndex.get(new ByteArrayWrapper(row.getKey()));
			resultList.set(i, columns.get(0).getValue());
		}
	}
	return resultList;
}
 

import me.prettyprint.cassandra.serializers.BytesArraySerializer; //导入依赖的package包/类
/** Creates and initializes a range query. */
private RangeSlicesQuery<byte[], String, byte[]> prepareRangeQuery(
		byte[] beginKey, byte[] endKey, boolean includeValue) {
	RangeSlicesQuery<byte[], String, byte[]> query = HFactory
			.createRangeSlicesQuery(keyspace, BytesArraySerializer.get(),
					StringSerializer.get(), BytesArraySerializer.get());
	query.setColumnFamily(CassandraStorageSystem.COLUMN_FAMILY_NAME);
	query.setColumnNames(CassandraStorageSystem.COLUMN_NAME);
	if (!includeValue) {
		query.setReturnKeysOnly();
	}
	query.setRowCount(MAX_SCAN_ROWS);
	query.setKeys(beginKey, endKey);
	return query;
}
 

import me.prettyprint.cassandra.serializers.BytesArraySerializer; //导入依赖的package包/类
@Override
  public StatefulHashMap<Utf8, T> fromByteBuffer(ByteBuffer byteBuffer) {
    if (byteBuffer == null) {
      return null;
    }
    StatefulHashMap<Utf8, T> map = new StatefulHashMap<Utf8, T>();
int i = 0;
    while (true) {
      Utf8 key = null;
      T value = null;
      try {
        int n = IntegerSerializer.get().fromByteBuffer(byteBuffer);
        byte[] bytes = new byte[n];
        byteBuffer.get(bytes, 0, n);
        key = Utf8Serializer.get().fromByteBuffer( BytesArraySerializer.get().toByteBuffer(bytes) );

        if (size > 0) {
          value = valueSerializer.fromByteBuffer(byteBuffer);
        }
        else {
          n = IntegerSerializer.get().fromByteBuffer(byteBuffer);
          bytes = new byte[n];
          byteBuffer.get(bytes, 0, n);
          value = valueSerializer.fromByteBuffer( BytesArraySerializer.get().toByteBuffer(bytes) );
        }
      } catch (BufferUnderflowException e) {
        break;
      }
      if (key == null) {
        break;
      }
      if (value == null) {
        break;
      }
      map.put(key, value);
    }
    return map;
  }
 

import me.prettyprint.cassandra.serializers.BytesArraySerializer; //导入依赖的package包/类
@Override
  public GenericArray<T> fromByteBuffer(ByteBuffer byteBuffer) {
    if (byteBuffer == null) {
      return null;
    }
    GenericArray<T> array = new ListGenericArray<T>(elementSchema);
int i = 0;
    while (true) {
      T element = null;
      try {
        if (size > 0) {
          element = elementSerializer.fromByteBuffer(byteBuffer);
        }
        else {
          int n = IntegerSerializer.get().fromByteBuffer(byteBuffer);
          byte[] bytes = new byte[n];
          byteBuffer.get(bytes, 0, n);
          element = elementSerializer.fromByteBuffer( BytesArraySerializer.get().toByteBuffer(bytes) );
        }
      } catch (BufferUnderflowException e) {
        break;
      }
      if (element == null) {
        break;
      }
      array.add(element);
    }
    return array;
  }
 

import me.prettyprint.cassandra.serializers.BytesArraySerializer; //导入依赖的package包/类
@Inject
public CassandraDatastore(@Named("HOSTNAME") final String hostname,
		CassandraConfiguration cassandraConfiguration,
		HectorConfiguration configuration,
		KairosDataPointFactory kairosDataPointFactory) throws DatastoreException
{
	try
	{
		m_cassandraConfiguration = cassandraConfiguration;
		m_singleRowReadSize = m_cassandraConfiguration.getSingleRowReadSize();
		m_multiRowSize = m_cassandraConfiguration.getMultiRowSize();
		m_multiRowReadSize = m_cassandraConfiguration.getMultiRowReadSize();
		m_kairosDataPointFactory = kairosDataPointFactory;
		m_keyspaceName = m_cassandraConfiguration.getKeyspaceName();

		m_rowKeyCache = new DataCache<DataPointsRowKey>(m_cassandraConfiguration.getRowKeyCacheSize());
		m_metricNameCache = new DataCache<String>(m_cassandraConfiguration.getStringCacheSize());
		m_tagNameCache = new DataCache<String>(m_cassandraConfiguration.getStringCacheSize());
		m_tagValueCache = new DataCache<String>(m_cassandraConfiguration.getStringCacheSize());

		CassandraHostConfigurator hostConfig = configuration.getConfiguration();
		int threadCount = hostConfig.buildCassandraHosts().length + 3;

		m_cluster = HFactory.getOrCreateCluster("kairosdb-cluster",
				hostConfig, m_cassandraConfiguration.getCassandraAuthentication());

		KeyspaceDefinition keyspaceDef = m_cluster.describeKeyspace(m_keyspaceName);

		if (keyspaceDef == null)
		{
			createSchema(m_cassandraConfiguration.getReplicationFactor());
		}

		//set global consistency level
		ConfigurableConsistencyLevel confConsLevel = new ConfigurableConsistencyLevel();
		confConsLevel.setDefaultReadConsistencyLevel(m_cassandraConfiguration.getDataReadLevel().getHectorLevel());
		confConsLevel.setDefaultWriteConsistencyLevel(m_cassandraConfiguration.getDataWriteLevel().getHectorLevel());

		//create keyspace instance with specified consistency
		m_keyspace = HFactory.createKeyspace(m_keyspaceName, m_cluster, confConsLevel);

		ReentrantLock mutatorLock = new ReentrantLock();

		m_dataPointWriteBuffer = new WriteBuffer<DataPointsRowKey, Integer, byte[]>(
				m_keyspace, CF_DATA_POINTS, m_cassandraConfiguration.getWriteDelay(),
				m_cassandraConfiguration.getMaxWriteSize(),
				DATA_POINTS_ROW_KEY_SERIALIZER,
				IntegerSerializer.get(),
				BytesArraySerializer.get(),
				createWriteBufferStats(CF_DATA_POINTS, hostname),
				mutatorLock, threadCount,
				m_cassandraConfiguration.getWriteBufferJobQueueSize());

		m_rowKeyWriteBuffer = new WriteBuffer<String, DataPointsRowKey, String>(
				m_keyspace, CF_ROW_KEY_INDEX, m_cassandraConfiguration.getWriteDelay(),
				m_cassandraConfiguration.getMaxWriteSize(),
				StringSerializer.get(),
				DATA_POINTS_ROW_KEY_SERIALIZER,
				StringSerializer.get(),
				createWriteBufferStats(CF_ROW_KEY_INDEX, hostname),
				mutatorLock, threadCount,
				m_cassandraConfiguration.getWriteBufferJobQueueSize());

		m_stringIndexWriteBuffer = new WriteBuffer<String, String, String>(
				m_keyspace, CF_STRING_INDEX,
				m_cassandraConfiguration.getWriteDelay(),
				m_cassandraConfiguration.getMaxWriteSize(),
				StringSerializer.get(),
				StringSerializer.get(),
				StringSerializer.get(),
				createWriteBufferStats(CF_STRING_INDEX, hostname),
				mutatorLock, threadCount,
				m_cassandraConfiguration.getWriteBufferJobQueueSize());
	}
	catch (HectorException e)
	{
		throw new DatastoreException(e);
	}
}
 

import me.prettyprint.cassandra.serializers.BytesArraySerializer; //导入依赖的package包/类
public ByteArrayRowSerializer(Serializer<K> rowKeySerializer, Serializer<T> topKeySerializer) {
    super(rowKeySerializer, topKeySerializer, new BytesArraySerializer());
}
 

import me.prettyprint.cassandra.serializers.BytesArraySerializer; //导入依赖的package包/类
@Override public TestWideEntity fromByteBuffer(ByteBuffer byteBuffer) {
    TestWideEntity entity = new TestWideEntity();
    SerializationUtils.deserialize(BytesArraySerializer.get().fromByteBuffer(byteBuffer), entity);

    return entity;
}
 

import me.prettyprint.cassandra.serializers.BytesArraySerializer; //导入依赖的package包/类
@Override public TestNonorganicWideEntityA fromByteBuffer(ByteBuffer byteBuffer) {
    TestNonorganicWideEntityA entity = new TestNonorganicWideEntityA();
    SerializationUtils.deserialize(BytesArraySerializer.get().fromByteBuffer(byteBuffer), entity);

    return entity;
}
 

import me.prettyprint.cassandra.serializers.BytesArraySerializer; //导入依赖的package包/类
@Override public TestPartitionedEntity fromByteBuffer(ByteBuffer byteBuffer) {
    TestPartitionedEntity entity = new TestPartitionedEntity();
    SerializationUtils.deserialize(BytesArraySerializer.get().fromByteBuffer(byteBuffer), entity);

    return entity;
}
 

import me.prettyprint.cassandra.serializers.BytesArraySerializer; //导入依赖的package包/类
@Override public TestNonorganicWideEntityB fromByteBuffer(ByteBuffer byteBuffer) {
    TestNonorganicWideEntityB entity = new TestNonorganicWideEntityB();
    SerializationUtils.deserialize(BytesArraySerializer.get().fromByteBuffer(byteBuffer), entity);

    return entity;
}