HRavenStatsWriteService implements StatsWriteService { static Configuration initHBaseConfiguration(Configuration jobConf) { Configuration hbaseConf = null; String hravenDir = System.getenv(HRAVEN_HBASE_CONF_DIR_ENV); if (hravenDir != null && !hravenDir.isEmpty()) { hbaseConf = new Configuration(false); Path hbaseConfFile = new Path(hravenDir, HRAVEN_HBASE_CONF_FILE); LOG.info("Loading hRaven HBase configuration from " + hbaseConfFile.toString()); hbaseConf.addResource(hbaseConfFile); } else { hbaseConf = HBaseConfiguration.create(); String hbaseZKQuorum = jobConf.get(HRAVEN_ZOOKEEPER_QUORUM); if (hbaseZKQuorum != null && !hbaseZKQuorum.isEmpty()) { try { LOG.info("Applying hRaven Zookeeper quorum information from job conf: " + hbaseZKQuorum); ZKUtil.applyClusterKeyToConf(hbaseConf, hbaseZKQuorum); } catch (IOException ioe) { throw new RuntimeException("Invalid cluster configuration for " + HRAVEN_ZOOKEEPER_QUORUM + " (\"" + hbaseZKQuorum + "\")"); } } else if (LOG.isDebugEnabled()) { LOG.debug("No cluster configuration found for " + HRAVEN_ZOOKEEPER_QUORUM + ", continuing with default"); } } return hbaseConf; } HRavenStatsWriteService(); void shutdown(); @Override void initWriteService(Properties properties); String getWorkflowId(); @Override void sendDagNodeNameMap(String workflowId,
    Map dagNodeMap); @SuppressWarnings("unchecked") @Override void pushEvent(String workflowId, Event event); static final String HRAVEN_HBASE_CONF_DIR_ENV; static final String HRAVEN_HBASE_CONF_FILE; static final String HRAVEN_ZOOKEEPER_QUORUM; }
@Test public void testInitHBaseConfiguration() { Configuration tmp = new Configuration(); String zkServers = "zk1,zk2,zk3"; String zkPort = "8181"; String parentZnode = "/hraven-test"; tmp.set(HConstants.ZOOKEEPER_QUORUM, zkServers); tmp.set(HConstants.ZOOKEEPER_CLIENT_PORT, zkPort); tmp.set(HConstants.ZOOKEEPER_ZNODE_PARENT, parentZnode); String zkConfigString = ZKUtil.getZooKeeperClusterKey(tmp); Configuration jobConf = new Configuration(); jobConf.set(HRavenStatsWriteService.HRAVEN_ZOOKEEPER_QUORUM, zkConfigString); Configuration hravenConfig = HRavenStatsWriteService.initHBaseConfiguration(jobConf); assertEquals(zkServers, hravenConfig.get(HConstants.ZOOKEEPER_QUORUM)); assertEquals(zkPort, hravenConfig.get(HConstants.ZOOKEEPER_CLIENT_PORT)); assertEquals(parentZnode, hravenConfig.get(HConstants.ZOOKEEPER_ZNODE_PARENT)); String zkServers2 = "zk4,zk5,zk6"; String zkPort2 = "9191"; String parentZnode2 = "/hraven-test2"; tmp = new Configuration(); tmp.set(HConstants.ZOOKEEPER_QUORUM, zkServers2); tmp.set(HConstants.ZOOKEEPER_CLIENT_PORT, zkPort2); tmp.set(HConstants.ZOOKEEPER_ZNODE_PARENT, parentZnode2); String zkConfigString2 = ZKUtil.getZooKeeperClusterKey(tmp); hravenConfig.set(HRavenStatsWriteService.HRAVEN_ZOOKEEPER_QUORUM, zkConfigString2); Configuration hravenConfig2 = HRavenStatsWriteService.initHBaseConfiguration(hravenConfig); assertEquals(zkServers2, hravenConfig2.get(HConstants.ZOOKEEPER_QUORUM)); assertEquals(zkPort2, hravenConfig2.get(HConstants.ZOOKEEPER_CLIENT_PORT)); assertEquals(parentZnode2, hravenConfig2.get(HConstants.ZOOKEEPER_ZNODE_PARENT)); }