HBaseUtils { public static Configuration getHBaseConfiguration(Config config) throws IOException { Configuration hbaseConfiguration = HBaseConfiguration.create(); if (config.hasPath(ZK_QUORUM_PROPERTY)) { String zkQuorum = config.getString(ZK_QUORUM_PROPERTY); hbaseConfiguration.set(HConstants.ZOOKEEPER_QUORUM, zkQuorum); } LOG.debug("HBase:: Using ZK quorum: {}", hbaseConfiguration.get(HConstants.ZOOKEEPER_QUORUM)); LOG.debug("HBase:: Using security: {}", hbaseConfiguration.get("hadoop.security.authentication")); if (config.hasPath(HBASE_PASSTHRU_PREFIX)) { Config hbaseConfigs = config.getConfig(HBASE_PASSTHRU_PREFIX); for (Map.Entry<String, ConfigValue> entry : hbaseConfigs.entrySet()) { String param = entry.getKey(); String value = null; switch (entry.getValue().valueType()) { case STRING: value = (String) entry.getValue().unwrapped(); break; default: LOG.warn("Only string parameters currently " + "supported, auto-converting to String [{}]", param); value = entry.getValue().unwrapped().toString(); } if (value != null) { hbaseConfiguration.set(param, value); } } } return hbaseConfiguration; }  synchronized static Connection getConnection(Config config); static Configuration getHBaseConfiguration(Config config); static int batchSizeFor(Config config); static TableName tableInfoFor(Config config); static HBaseSerde getSerde(Config config); static byte[] rowKeySeparatorFor(Config config); static List<String> rowKeyFor(Config config); static Map<String, HBaseSerde.ColumnDef> columnsFor(Config config); static StructType buildSchema(Map<String, HBaseSerde.ColumnDef> columnDefinitions); static Scan mergeRangeScans(List<Scan> rangeScans); static byte[] exclusiveStopRow(byte[] startRow); static Validations getValidations(); static final String ZK_QUORUM_PROPERTY; static final String TABLE_NAME_PROPERTY; static final String ROWKEY_PROPERTY; static final String KEY_SEPARATOR; static final String COLUMNS_PROPERTY; static final String SERDE_PROPERTY; static final String HBASE_PASSTHRU_PREFIX; static final String HBASE_BATCH_SIZE; static final int DEFAULT_HBASE_BATCH_SIZE; static final String DEFAULT_SERDE_PROPERTY; static final String DEFAULT_KEY_SEPARATOR; }
@Test public void testPassthroughHBaseOptions() throws IOException { Config goodConfig = ConfigUtils.configFromResource("/hbase/hbase-output-with-hbase-config.conf").getConfig("output"); List<ValidationResult> results = Validator.validate(new HBaseOutput(), goodConfig); assertFalse("Good config with HBase options should validate", ValidationUtils.hasValidationFailures(results)); Configuration hBaseConfiguration = HBaseUtils.getHBaseConfiguration(goodConfig); assertEquals("HBase configuration should have custom retries", 5, Integer.parseInt(hBaseConfiguration.get("hbase.client.retries.number"))); assertEquals("HBase configuration should have custom timeout", 30000, Integer.parseInt(hBaseConfiguration.get("hbase.client.operation.timeout"))); }