import org.apache.kudu.client.KuduException; //导入依赖的package包/类
public Map<SchemaTableName, KuduTableHandle> getTables(KuduClient kuduClient)
    {
        Map<SchemaTableName, KuduTableHandle> tables = null;
        ImmutableMap.Builder<SchemaTableName, KuduTableHandle> tablesBuilder = ImmutableMap.builder();
//        ImmutableMap.Builder<SchemaTableName, List<ColumnMetadata>> tableColumnsBuilder = ImmutableMap.builder();
        List<String> listTable = null;
        try {
            listTable = kuduClient.getTablesList().getTablesList();
        }
        catch (KuduException e) {
            e.printStackTrace();
        }

        for (String table : listTable) {
            SchemaTableName schemaTableName = new SchemaTableName(PRESTO_KUDU_SCHEMA, table);
            tablesBuilder.put(schemaTableName, new KuduTableHandle(schemaTableName));
        }

        tables = tablesBuilder.build();

        return tables;
    }
 

import org.apache.kudu.client.KuduException; //导入依赖的package包/类
@Override
public boolean advanceNextPosition()
{
    try {
        if (results != null && results.hasNext()) {
            result = results.next();
            return true;
        }
        else {
            if (kuduScanner.hasMoreRows()) {
                results = kuduScanner.nextRows();
                if (results.hasNext()) {
                    result = results.next();
                    return true;
                }
            }
        }
    }
    catch (KuduException e) {
        logger.error(e, e.getMessage());
    }
    return false;
}
 

import org.apache.kudu.client.KuduException; //导入依赖的package包/类
private synchronized KuduTable connectToTable() throws KuduException {
  if (client == null) {
    LOG.info("Connecting to Kudu");

    String masterAddresses = config.getString(CONNECTION_CONFIG_NAME);

    client = new KuduClient.KuduClientBuilder(masterAddresses).build();
    session = client.newSession();
    session.setFlushMode(FlushMode.AUTO_FLUSH_BACKGROUND);
    session.setMutationBufferSpace(10000);
    session.setIgnoreAllDuplicateRows(isInsertIgnore());

    LOG.info("Connection to Kudu established");
  }

  String tableName = config.getString(TABLE_CONFIG_NAME);
  KuduTable table = getTable(tableName);

  return table;
}
 

import org.apache.kudu.client.KuduException; //导入依赖的package包/类
@Test
public void testConnectionFailure() throws Exception{
  // Mock connection refused
  PowerMockito.stub(
      PowerMockito.method(AsyncKuduClient.class, "getTablesList"))
      .toThrow(PowerMockito.mock(KuduException.class));

  ProcessorRunner runner = setProcessorRunner(tableName);

  try {
    List<Stage.ConfigIssue> issues = runner.runValidateConfigs();
    Assert.assertEquals(1, issues.size());
  } catch (StageException e) {
    Assert.fail("should not throw StageException");
  }
}
 

import org.apache.kudu.client.KuduException; //导入依赖的package包/类
/**
 * If table name template is not EL, we access Kudu and check if the table exists.
 * This test checks when Kudu's tableExists() method returns false.
 * @throws Exception
 */
@Test
public void testTableDoesNotExistNoEL() throws Exception{
  // Mock table doesn't exist in Kudu.
  PowerMockito.stub(
      PowerMockito.method(AsyncKuduClient.class, "tableExists"))
      .toThrow(PowerMockito.mock(KuduException.class));
  ProcessorRunner runner = setProcessorRunner(tableName);

  try {
    List<Stage.ConfigIssue> issues = runner.runValidateConfigs();
    Assert.assertEquals(1, issues.size());
  } catch (StageException e){
    Assert.fail();
  }
}
 

import org.apache.kudu.client.KuduException; //导入依赖的package包/类
@Test
public void testConnectionFailure() throws Exception{
  // Mock connection refused
  PowerMockito.stub(
      PowerMockito.method(KuduClient.class, "getTablesList"))
      .toThrow(PowerMockito.mock(KuduException.class));

  TargetRunner targetRunner = setTargetRunner(tableName, KuduOperationType.INSERT, UnsupportedOperationAction.DISCARD);

  try {
    List<Stage.ConfigIssue> issues = targetRunner.runValidateConfigs();
    Assert.assertEquals(1, issues.size());
  } catch (StageException e) {
    Assert.fail("should not throw StageException");
  }
}
 

import org.apache.kudu.client.KuduException; //导入依赖的package包/类
@Override
public boolean validateObject(PooledObject<KuduClient> pooledKuduClient)
{
    final KuduClient kuduClient = pooledKuduClient.getObject();
    try {
        kuduClient.listTabletServers();
        return true;
    }
    catch (KuduException e) {
        return false;
    }
}
 

import org.apache.kudu.client.KuduException; //导入依赖的package包/类
public KuduTable openTable(KuduClient client, final String name)

    {
        KuduTable kuduTable = null;
        try {
            kuduTable = client.openTable(name);
        }
        catch (KuduException e) {
            log.error(e, e.getMessage());
        }
        return kuduTable;
    }
 

import org.apache.kudu.client.KuduException; //导入依赖的package包/类
public void dropTable(KuduClient kuduClient, String tableName)
{
    try {
        kuduClient.deleteTable(tableName);
    }
    catch (KuduException e) {
        log.error(e, e.getMessage());
    }
}
 

import org.apache.kudu.client.KuduException; //导入依赖的package包/类
@Override
public void close()
{
    // TODO: 18/04/2017 如果有资源申请,在这个地方需要关闭
    try {
        if (kuduScanner != null) {
            kuduScanner.close();
        }
        kuduClientManager.close(this.kuduClient);
    }
    catch (KuduException e) {
        logger.error(e, e.getMessage());
    }
}
 

import org.apache.kudu.client.KuduException; //导入依赖的package包/类
private Row resultAsRow(RowResult result, KuduTable table) throws KuduException {
  List<Object> values = Lists.newArrayList();

  for (ColumnSchema columnSchema : table.getSchema().getColumns()) {
    String columnName = columnSchema.getName();

    if (result.isNull(columnName)) {
      values.add(null);
      continue;
    }

    switch (columnSchema.getType()) {
      case DOUBLE:
        values.add(result.getDouble(columnName));
        break;
      case FLOAT:
        values.add(result.getFloat(columnName));
        break;
      case INT32:
        values.add(result.getInt(columnName));
        break;
      case INT64:
        values.add(result.getLong(columnName));
        break;
      case STRING:
        values.add(result.getString(columnName));
        break;
      case BOOL:
        values.add(result.getBoolean(columnName));
        break;
      default:
        throw new RuntimeException("Unsupported Kudu column type: " + columnSchema.getType());
    }
  }

  Row row = new RowWithSchema(getTableSchema(table), values.toArray());

  return row;
}
 

import org.apache.kudu.client.KuduException; //导入依赖的package包/类
private synchronized KuduTable getTable(String tableName) throws KuduException {
  if (tables == null) {
    tables = Maps.newHashMap();
  }

  if (tables.containsKey(tableName)) {
    return tables.get(tableName);
  }
  else {
    KuduTable table = client.openTable(tableName);
    tables.put(tableName, table);
    return table;
  }
}
 

import org.apache.kudu.client.KuduException; //导入依赖的package包/类
private synchronized StructType getTableSchema(KuduTable table) throws KuduException {
  if (tableSchemas == null) {
    tableSchemas = Maps.newHashMap();
  }

  if (tableSchemas.containsKey(table.getName())) {
    return tableSchemas.get(table.getName());
  }
  else {
    StructType tableSchema = schemaFor(table);
    tableSchemas.put(table.getName(), tableSchema);
    return tableSchema;
  }
}
 

import org.apache.kudu.client.KuduException; //导入依赖的package包/类
/**
 * Attempts to insert the provided tagset and ID. Returns {@code true} if the
 * write was successful, or {@code false} if the write failed due to a tagset
 * with the same ID already existing in the table.
 *
 * @param tagset the tagset to insert
 * @param id     the ID to insert the tagset with
 * @return whether the write succeeded
 */
private Deferred<Boolean> insertTagset(final SerializedTagset tagset, final int id) throws KuduException {
  final class InsertTagsetCB implements Callback<Deferred<Boolean>, OperationResponse> {
    @Override
    public Deferred<Boolean> call(OperationResponse response) {
      if (response.hasRowError()) {
        if (response.getRowError().getErrorStatus().isAlreadyPresent()) {
          LOG.info("Attempted to insert duplicate tagset; id: {}, tagset: {}", id, tagset);
          // TODO: Consider adding a backoff with jitter before attempting
          //       the insert again (if the lookup fails).
          return Deferred.fromResult(false);
        }
        return Deferred.fromError(new RuntimeException(
            String.format("Unable to insert tagset; id: %s, tagset: %s, error: %s",
                          id, tagset, response.getRowError())));
      } else {
        return Deferred.fromResult(true);
      }
    }
    @Override
    public String toString() {
      return MoreObjects.toStringHelper(this).toString();
    }
  }

  LOG.debug("Inserting tagset; id: {}, tags: {}", id, tagset);
  final AsyncKuduSession session = client.newSession();
  try {
    // We don't have to handle PleaseThrottleException because we are only
    // inserting a single row.
    final Insert insert = tagsetsTable.newInsert();
    insert.getRow().addInt(Tables.TAGSETS_ID_INDEX, id);
    insert.getRow().addBinary(Tables.TAGSETS_TAGSET_INDEX, tagset.getBytes());
    return session.apply(insert).addCallbackDeferring(new InsertTagsetCB());
  } finally {
    session.close();
  }
}
 

import org.apache.kudu.client.KuduException; //导入依赖的package包/类
public static void createTestTable(String tableName, KuduClient client) throws Exception
{
  List<String> rangeKeys = new ArrayList<>();
  rangeKeys.add("introwkey");
  List<String> hashPartitions = new ArrayList<>();
  hashPartitions.add("stringrowkey");
  hashPartitions.add("timestamprowkey");
  CreateTableOptions thisTableOptions = new CreateTableOptions()
      .setNumReplicas(1)
      .addHashPartitions(hashPartitions,HASH_BUCKETS_SIZE_FOR_ALL_HASH_COL)
      .setRangePartitionColumns(rangeKeys);
  int stepsize = Integer.MAX_VALUE / SPLIT_COUNT_FOR_INT_ROW_KEY;
  int splitBoundary = stepsize;
  Schema schema = buildSchemaForUnitTestsTable();
  for ( int i = 0; i < SPLIT_COUNT_FOR_INT_ROW_KEY; i++) {
    PartialRow splitRowBoundary = schema.newPartialRow();
    splitRowBoundary.addInt("introwkey",splitBoundary);
    thisTableOptions = thisTableOptions.addSplitRow(splitRowBoundary);
    splitBoundary += stepsize;
  }
  try {
    client.createTable(tableName, schema,thisTableOptions);
  } catch (KuduException e) {
    LOG.error("Error while creating table for unit tests " + e.getMessage(), e);
    throw e;
  }

}
 

import org.apache.kudu.client.KuduException; //导入依赖的package包/类
public KuduTarget(KuduConfigBean configBean) {
  this.configBean = configBean;
  this.kuduMaster = Strings.nullToEmpty(configBean.kuduMaster).trim();
  this.tableNameTemplate = Strings.nullToEmpty(configBean.tableNameTemplate).trim();
  this.fieldMappingConfigs = configBean.fieldMappingConfigs == null
      ? Collections.<KuduFieldMappingConfig>emptyList()
      : configBean.fieldMappingConfigs;
  this.defaultOperation = configBean.defaultOperation;

  CacheBuilder cacheBuilder = CacheBuilder.newBuilder()
      .maximumSize(tableCacheSize)
      .expireAfterAccess(1, TimeUnit.HOURS);

  if(LOG.isDebugEnabled()) {
    cacheBuilder.recordStats();
  }

  kuduTables = cacheBuilder.build(new CacheLoader<String, KuduTable>() {
        @Override
        public KuduTable load(String tableName) throws KuduException {
          return kuduClient.openTable(tableName);
        }
      });

  cacheCleaner = new CacheCleaner(kuduTables, "KuduTarget", 10 * 60 * 1000);

  recordConverters = cacheBuilder.build(new CacheLoader<KuduTable, KuduRecordConverter>() {
    @Override
    public KuduRecordConverter load(KuduTable table) throws KuduException {
      return createKuduRecordConverter(table);
    }
  });
  recordBuilderCacheCleaner = new CacheCleaner(recordConverters, "KuduTarget KuduRecordConverter", 10 * 60 * 1000);
}
 

import org.apache.kudu.client.KuduException; //导入依赖的package包/类
@Test
public void testCheckConnection() throws Exception{
  PowerMockito.stub(
      PowerMockito.method(AsyncKuduClient.class, "getTablesList"))
      .toThrow(PowerMockito.mock(KuduException.class));

  List<Stage.ConfigIssue> issues = new ArrayList<>();
  Context context = getContext();
  KuduUtils.checkConnection(client, context, KUDU_MASTER, issues);
  Assert.assertEquals(1, issues.size());
  Stage.ConfigIssue issue = issues.get(0);
  Assert.assertTrue(issues.get(0).toString().contains(Errors.KUDU_00.name()));
}
 

import org.apache.kudu.client.KuduException; //导入依赖的package包/类
/**
 * When tableNameTemplate contains EL, it checks if the table exists in Kudu
 * for every single records. This test checks error records when the table
 * does not exist in Kudu.
 * @throws Exception
 */
@Test
public void testTableDoesNotExistWithEL() throws Exception{
  // Mock KuduClient.openTable() to throw KuduException
  PowerMockito.stub(PowerMockito.method(KuduClient.class, "openTable")).toThrow(PowerMockito.mock(KuduException.class));

  TargetRunner targetRunner = setTargetRunner(
      "${record:attribute('tableName')}",
      KuduOperationType.INSERT,
      UnsupportedOperationAction.DISCARD
  );

  Record record =  RecordCreator.create();
  LinkedHashMap<String, Field> field = new LinkedHashMap<>();
  field.put("key", Field.create(1));
  record.set(Field.createListMap(field));
  record.getHeader().setAttribute("tableName", "do_not_exist");
  targetRunner.runInit();

  try {
    targetRunner.runWrite(Collections.singletonList(record));
    Assert.assertEquals(1, targetRunner.getErrorRecords().size());
  } catch (StageException e){
    Assert.fail();
  }
  targetRunner.runDestroy();
}
 

import org.apache.kudu.client.KuduException; //导入依赖的package包/类
/**
 * Insert a tagset into the {@code tags} table.
 * @param id the tagset ID.
 * @param tagset the tagset.
 * @return The tagset ID.
 */
public Deferred<Integer> insertTagset(final int id, final SortedMap<String, String> tagset)
    throws KuduException {
  if (tagset.isEmpty()) { return Deferred.fromResult(id); }
  LOG.debug("Inserting tags; tagsetID: {}, tags: {}", id, tagset);
  final AsyncKuduSession session = client.newSession();

  class InsertTagsetCB implements Callback<Deferred<Integer>, List<OperationResponse>> {
    @Override
    public Deferred<Integer> call(List<OperationResponse> responses) {
      try {
        for (OperationResponse response : responses) {
          if (response.hasRowError()) {
            return Deferred.fromError(new RuntimeException(
                String.format("Unable to insert tag: %s", response.getRowError())));
          }
        }
        return Deferred.fromResult(id);
      } finally {
        session.close();
      }
    }
    @Override
    public String toString() {
      return MoreObjects.toStringHelper(this)
                        .add("id", id)
                        .add("tags", tagset)
                        .toString();
    }
  }

  if (tagset.size() > 1000) {
    session.setMutationBufferSpace(tagset.size());
  }
  session.setMutationBufferLowWatermark(1.0f);

  // buffer all of the tags into the session, and ensure that we don't get
  // a PleaseThrottleException. In practice the number of tags should be
  // small.
  session.setMutationBufferSpace(tagset.size());
  session.setMutationBufferLowWatermark(1.0f);
  session.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
  for (Map.Entry<String, String> tag : tagset.entrySet()) {
    Insert insert = table.newInsert();
    // TODO: check with JD that if the inserts below fail, the error will
    // also be returned in the flush call.
    insert.getRow().addString(Tables.TAGS_KEY_INDEX, tag.getKey());
    insert.getRow().addString(Tables.TAGS_VALUE_INDEX, tag.getValue());
    insert.getRow().addInt(Tables.TAGS_TAGSET_ID_INDEX, id);
    session.apply(insert);
  }
  return session.flush().addCallbackDeferring(new InsertTagsetCB());
}