import me.prettyprint.hector.api.exceptions.HectorException; //导入依赖的package包/类
/** {@inheritDoc} */
@Override
protected List<byte[]> doBatchGet(List<byte[]> keys)
		throws StorageException {
	try {
		List<byte[]> result = new ArrayList<byte[]>();
		for (int i = 0; i < keys.size(); i += MAX_BATCH_ROWS) {
			List<byte[]> subKeyList = keys.subList(i,
					Math.min(i + MAX_BATCH_ROWS, keys.size()));
			result.addAll(batchGetSubList(subKeyList));
		}
		return result;
	} catch (HectorException e) {
		handleHectorException(e);
		return null;
	}
}
 

import me.prettyprint.hector.api.exceptions.HectorException; //导入依赖的package包/类
/** {@inheritDoc} */
@Override
protected void doBatchPut(PairList<byte[], byte[]> keysValues)
		throws StorageException {
	try {
		Mutator<byte[]> mutator = template.createMutator();
		for (int i = 0; i < keysValues.size(); ++i) {
			HColumn<String, byte[]> column = HFactory.createColumn(
					CassandraStorageSystem.COLUMN_NAME,
					keysValues.getSecond(i));
			mutator.addInsertion(keysValues.getFirst(i),
					CassandraStorageSystem.COLUMN_FAMILY_NAME, column);

			if ((i + 1) % MAX_BATCH_ROWS == 0) {
				mutator.execute();
			}
		}
		mutator.execute();
	} catch (HectorException e) {
		handleHectorException(e);
	}
}
 

import me.prettyprint.hector.api.exceptions.HectorException; //导入依赖的package包/类
/** {@inheritDoc} */
@Override
protected void doBatchRemove(List<byte[]> keys) throws StorageException {
	try {
		Mutator<byte[]> mutator = template.createMutator();
		int index = 0;
		for (byte[] key : keys) {
			mutator.addDeletion(key,
					CassandraStorageSystem.COLUMN_FAMILY_NAME);
			index += 1;
			if (index % MAX_BATCH_ROWS == 0) {
				mutator.execute();
			}
		}
		mutator.execute();
	} catch (HectorException e) {
		handleHectorException(e);
	}
}
 

import me.prettyprint.hector.api.exceptions.HectorException; //导入依赖的package包/类
/** {@inheritDoc} */
@Override
protected void doScan(byte[] beginKey, byte[] endKey,
		IKeyValueCallback callback, boolean includeValue)
		throws StorageException {
	try {
		RangeSlicesQuery<byte[], String, byte[]> query = prepareRangeQuery(
				beginKey, endKey, includeValue);
		byte[] lastKey = null;
		while (true) {
			if (lastKey != null) {
				query.setKeys(lastKey, endKey);
			}

			OrderedRows<byte[], String, byte[]> rows = query.execute()
					.get();
			lastKey = reportScanData(rows, callback, endKey, includeValue);

			if (rows.getCount() < MAX_SCAN_ROWS) {
				break;
			}
		}
	} catch (HectorException e) {
		handleHectorException(e);
	}
}
 

import me.prettyprint.hector.api.exceptions.HectorException; //导入依赖的package包/类
public CassandraResultSet<K, String> readAllColumns(K key) throws HectorException {
	try {
		if (maxColumnCount > 0) {
			return readColumnSlice(key, null, null, false, maxColumnCount);
		} else {
			ColumnFamilyResult<K, String> queriedColumns = getColumnFamily().queryColumns(key);
			if (isClientAdapterDebugMessagesEnabled) {
				log.info("Row retrieved from Cassandra. Exec Time (micro-sec) = " +
						queriedColumns.getExecutionTimeMicro() +
						", Host used = " + queriedColumns.getHostUsed() + ", Key = " + key);
			}
			return new HectorResultSet<K, String>(queriedColumns);
		}

	} catch (HectorException e) {
		log.debug("HecubaClientManager error while reading key " + key.toString());
		if (log.isDebugEnabled()) {
			log.debug("Caught Exception", e);

			// lets see whether we have any issues with the downed nodes.
			logDownedHosts();
		}
		throw e;
	}
}
 

import me.prettyprint.hector.api.exceptions.HectorException; //导入依赖的package包/类
public CassandraResultSet<K, String> readColumns(K key, List<String> columns) throws HectorException {
	ColumnFamilyResult<K, String> queriedColumns;
	try {
		queriedColumns = getColumnFamily().queryColumns(key, columns);
		if (isClientAdapterDebugMessagesEnabled) {
			log.info(columns.size() + " columns retrieved from Cassandra. Exec Time = " +
					queriedColumns.getExecutionTimeMicro() + ", Host used = " +
					queriedColumns.getHostUsed());
		}
		return new HectorResultSet<K, String>(queriedColumns);
	} catch (HectorException e) {
		log.error("HecubaClientManager error while retrieving " + columns.size() + " columns for key " +
				key.toString());
		if (log.isDebugEnabled()) {
			log.debug("Caught Exception", e);
		}
		throw e;
	}
}
 

import me.prettyprint.hector.api.exceptions.HectorException; //导入依赖的package包/类
/**
 * Deletes a user by userName.
 */
@Override
public void doDeleteUser(String userName) throws UserStoreException {

    Mutator<Composite> mutator = HFactory.createMutator(keyspace, CompositeSerializer.get());
    String[] roles = doGetExternalRoleListOfUser(userName, "");
    for (String role : roles) {
        Composite key = new Composite();
        key.addComponent(role, stringSerializer);
        key.addComponent(tenantIdString, stringSerializer);
        ColumnFamilyTemplate<Composite, String> userCFTemplate = new ThriftColumnFamilyTemplate<Composite, String>(
                keyspace, CFConstants.UM_ROLE_USER_INDEX, CompositeSerializer.get(), StringSerializer.get());
        try {
            userCFTemplate.deleteColumn(key, userName);
        } catch (HectorException e) {
            log.error("Error during deletion ", e);
        }
    }

    Composite userKey = new Composite();
    userKey.addComponent(userName, stringSerializer);
    userKey.addComponent(tenantIdString, stringSerializer);
    mutator.addDeletion(userKey, CFConstants.UM_USER_ROLE, null, CompositeSerializer.get());
    mutator.addDeletion(userKey, CFConstants.UM_USER, null, CompositeSerializer.get());
    mutator.execute();

    if (log.isDebugEnabled()) {
        log.debug("Deleted user " + userName + " successfully");
    }
}
 

import me.prettyprint.hector.api.exceptions.HectorException; //导入依赖的package包/类
@Override
public void doUpdateCredentialByAdmin(String userName, Object newCredential) throws UserStoreException {
    if (!checkUserPasswordValid(newCredential)) {
        throw new UserStoreException(
                "Credential not valid. Credential must be a non null string with following format, "
                        + realmConfig.getUserStoreProperty(UserCoreConstants.RealmConfig.PROPERTY_JAVA_REG_EX));

    }

    String saltValue = null;
    if (TRUE.equalsIgnoreCase(realmConfig.getUserStoreProperties().get(JDBCRealmConstants.STORE_SALTED_PASSWORDS))) {
        saltValue = Util.getSaltValue();
    }
    String password = Util.preparePassword((String) newCredential, saltValue);
    Mutator<Composite> mutator = HFactory.createMutator(keyspace, CompositeSerializer.get());
    Composite key = new Composite();
    key.addComponent(userName, stringSerializer);
    key.addComponent(tenantIdString, stringSerializer);
    mutator.addInsertion(key, CFConstants.UM_USER,
            HFactory.createColumn(CFConstants.UM_SECRET, password, stringSerializer, stringSerializer));
    mutator.addInsertion(key, CFConstants.UM_USER,
            HFactory.createColumn(CFConstants.UM_SALT_VALUE, saltValue, stringSerializer, stringSerializer));
    try {
        mutator.execute();
        if (log.isDebugEnabled()) {
            log.debug("Changed password for user " + userName + "successfully");
        }
    } catch (HectorException e) {
        throw new UserStoreException("Change Password failed.", e);
    }
}
 

import me.prettyprint.hector.api.exceptions.HectorException; //导入依赖的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.hector.api.exceptions.HectorException; //导入依赖的package包/类
/** {@inheritDoc} */
@Override
protected byte[] doGet(byte[] key) throws StorageException {
	try {
		ColumnFamilyResult<byte[], String> res = template.queryColumns(key);
		return res.getByteArray(CassandraStorageSystem.COLUMN_NAME);
	} catch (HectorException e) {
		handleHectorException(e);
		return null;
	}
}
 

import me.prettyprint.hector.api.exceptions.HectorException; //导入依赖的package包/类
/** Handles a {@link HectorException} . */
private void handleHectorException(HectorException e)
		throws StorageException {
	if (e instanceof HTimedOutException) {
		throw new TimeoutException(e);
	} else if (e.getMessage().startsWith("All host pools marked down")) {
		throw new TimeoutException(e);
	}
	throw new StorageException(e);
}
 

import me.prettyprint.hector.api.exceptions.HectorException; //导入依赖的package包/类
/** {@inheritDoc} */
@Override
protected void doPut(byte[] key, byte[] value) throws StorageException {
	try {
		ColumnFamilyUpdater<byte[], String> updater = template
				.createUpdater(key);
		updater.setByteArray(CassandraStorageSystem.COLUMN_NAME, value);
		template.update(updater);
	} catch (HectorException e) {
		handleHectorException(e);
	}
}
 

import me.prettyprint.hector.api.exceptions.HectorException; //导入依赖的package包/类
/** {@inheritDoc} */
@Override
protected void doRemove(byte[] key) throws StorageException {
	try {
		template.deleteRow(key);
	} catch (HectorException e) {
		handleHectorException(e);
	}
}
 

import me.prettyprint.hector.api.exceptions.HectorException; //导入依赖的package包/类
private List<K> retrieveKeysFromSecondaryIndex(String columnName, String columnValue) {
	String secondaryIndexKey = getSecondaryIndexKey(columnName, columnValue);
	CassandraResultSet<String, K> columns;
	try {
		if (maxSiColumnCount > 0) {
			columns = readSiColumnSlice(secondaryIndexKey, false, maxSiColumnCount);
		} else {
			columns = new HectorResultSet<String, K>(secondaryIndexedColumnFamilyTemplate.queryColumns(
					secondaryIndexKey));
		}

		if (isClientAdapterDebugMessagesEnabled) {
			log.debug("Secondary Index Row for " + secondaryIndexKey + " retrieved from Cassandra. Exec Time = " +
					columns.getExecutionLatency() + ", Host used = " + columns.getHost());
		}

		if (columns != null && CollectionUtils.isNotEmpty(columns.getColumnNames())) {
			return new ArrayList<>(columns.getColumnNames());
		}
	} catch (HectorException e) {
		log.debug("HecubaClientManager error while retrieving secondary index " + getSecondaryIndexKey(columnName,
				columnValue));
		if (log.isDebugEnabled()) {
			log.debug("Caught Exception", e);

			// lets see whether we have any issues with the downed nodes.
			logDownedHosts();
		}
		throw e;
	}
	return null;
}
 

import me.prettyprint.hector.api.exceptions.HectorException; //导入依赖的package包/类
/**
 * Wraps "describe_thrift_version API call as this hits a static string in Cassandra. This is the most lightweight
 * way to assure that Hector is alive and talking to the cluster.
 *
 * @return true if we have a lit connection to the cluster.
 */
public boolean getIsCassandraAlive() {
    boolean isAlive = false;
    try {
        isAlive = cluster.describeThriftVersion() != null;
    }
    catch ( HectorException he ) {
        logger.error( "Could not communicate with Cassandra cluster", he );
    }
    return isAlive;
}
 

import me.prettyprint.hector.api.exceptions.HectorException; //导入依赖的package包/类
public static void process(FailoverConf conf, Logger logger, final VoidFailoverQuery query) throws HectorException {
    process(conf, logger, new FailoverQuery<Void>() {
        @Override public Void query() {
            query.query();

            return null;
        }
    });
}
 

import me.prettyprint.hector.api.exceptions.HectorException; //导入依赖的package包/类
/**
 * Get a string value.
 *
 * @return The string value; null if no value exists for the given key.
 */
public String get(final String key, final String columnName) throws HectorException {
  ColumnQuery<String, String, String> q = HFactory.createColumnQuery(keyspace,
      serializer, serializer, serializer);
  QueryResult<HColumn<String, String>> r = q.setKey(key).
      setName(columnName).
      setColumnFamily(columnFamilyName).
      execute();
  HColumn<String, String> c = r.get();
  return c != null ? c.getValue() : null;
}
 

import me.prettyprint.hector.api.exceptions.HectorException; //导入依赖的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.hector.api.exceptions.HectorException; //导入依赖的package包/类
/**
 * Adds the user to the user store.
 */
@Override
public void doAddUser(String userName, Object credential, String[] roleList, Map<String, String> claims,
                      String profileName, boolean requirePasswordChange) throws UserStoreException {

    String userId = UUID.randomUUID().toString();
    String saltValue = null;

    if (TRUE.equalsIgnoreCase(realmConfig.getUserStoreProperties().get(JDBCRealmConstants.STORE_SALTED_PASSWORDS))) {
        saltValue = Util.getSaltValue();
    }

    String password = Util.preparePassword((String) credential, saltValue);

    if (doCheckExistingUser(userName)) {

        String message = "User with credentials " + userName + "exists";

        UserStoreException userStoreException = new UserStoreException(message);
        log.error(message, userStoreException);
        throw userStoreException;
    } else {

        Mutator<Composite> mutator = HFactory.createMutator(keyspace, CompositeSerializer.get());

        Composite key = new Composite();
        key.addComponent(userName, stringSerializer);
        key.addComponent(tenantIdString, stringSerializer);

        // add user ID
        mutator.addInsertion(key, CFConstants.UM_USER,
                HFactory.createColumn(CFConstants.UM_USER_ID, userId, stringSerializer, stringSerializer));
        mutator.addInsertion(key, CFConstants.UM_USER,
                HFactory.createColumn(CFConstants.UM_USER_NAME, userName, stringSerializer, stringSerializer));
        mutator.addInsertion(key, CFConstants.UM_USER,
                HFactory.createColumn(CFConstants.UM_SECRET, password, stringSerializer, stringSerializer));
        mutator.addInsertion(key, CFConstants.UM_USER,
                HFactory.createColumn(CFConstants.UM_SALT_VALUE, saltValue, stringSerializer, stringSerializer));
        mutator.addInsertion(key, CFConstants.UM_USER, HFactory.createColumn(CFConstants.UM_REQUIRE_CHANGE_BOOLEAN,
                "false", stringSerializer, stringSerializer));
        mutator.addInsertion(key, CFConstants.UM_USER,
                HFactory.createColumn(CFConstants.UM_TENANT_ID, tenantIdString, stringSerializer, stringSerializer));
        mutator = addUserToRoleList(userName, roleList, mutator);

        if (claims != null) {
            mutator = addClaimsForUser(userId, claims, mutator);
        }

        try {
            mutator.execute();
            if (log.isDebugEnabled()) {
                log.debug("Added user " + userName + " successfully");
            }
        } catch (HectorException e) {
            // TODO- research and check how to identify cassandra failure
            // and handle it efficiently.
            throw new UserStoreException("Adding user failed.", e);
        }
        mutator.execute();

    }
}
 

import me.prettyprint.hector.api.exceptions.HectorException; //导入依赖的package包/类
/**
 * Update the user list mapped to a role.
 */
@Override
public void doUpdateUserListOfRole(String roleName, String[] deletedUsers, String[] newUsers)
        throws UserStoreException {

    Mutator<Composite> mutator = HFactory.createMutator(keyspace, CompositeSerializer.get());
    RoleContext ctx = createRoleContext(roleName);
    roleName = ctx.getRoleName();
    boolean isShared = ctx.isShared();
    if (!isShared) {
        //TODO TO BE Implemented
    }
    if (deletedUsers != null && deletedUsers.length > 0) {
        if (isShared) {
            //TODO TO BE Implemented
        } else {
            if (deletedUsers.length > 0) {
                Composite key = new Composite();
                key.addComponent(roleName, stringSerializer);
                key.addComponent(tenantIdString, stringSerializer);

                for (String user : deletedUsers) {

                    Composite userKey = new Composite();
                    userKey.addComponent(user, stringSerializer);
                    userKey.addComponent(tenantIdString, stringSerializer);

                    ColumnFamilyTemplate<Composite, String> userCFTemplate = new ThriftColumnFamilyTemplate<Composite, String>(
                            keyspace, CFConstants.UM_USER_ROLE, CompositeSerializer.get(), StringSerializer.get());
                    ColumnFamilyTemplate<Composite, String> roleCFTemplate = new ThriftColumnFamilyTemplate<Composite, String>(
                            keyspace, CFConstants.UM_ROLE_USER_INDEX, CompositeSerializer.get(),
                            StringSerializer.get());
                    try {
                        roleCFTemplate.deleteColumn(mutator, key, user);
                        userCFTemplate.deleteColumn(mutator, userKey, roleName);
                    } catch (HectorException e) {
                        log.error(e.getMessage(), e);
                        throw new UserStoreException("Error during the updating of a user's role list");
                    }
                }
            }

        }
    }
    // need to clear user roles cache upon roles update
    clearUserRolesCacheByTenant(this.tenantId);

    if (newUsers != null && newUsers.length > 0) {
        if (isShared) {
            //TODO TO BE Implemented
        } else {
            addRoleToUsersList(newUsers, roleName, mutator);
        }
    }
    mutator.execute();

}