@Test(expected = FlumeException.class) public void testZKQuorumIncorrectPorts() throws Exception { initContextForSimpleHbaseEventSerializer(); Context tmpContext = new Context(ctx.getParameters()); String zkQuorum = "zk1.flume.apache.org:3345, zk2.flume.apache.org:3342, " + "zk3.flume.apache.org:3342"; tmpContext.put("batchSize", "2"); tmpContext.put(HBaseSinkConfigurationConstants.ZK_QUORUM, zkQuorum); tmpContext.put(HBaseSinkConfigurationConstants.ZK_ZNODE_PARENT, conf.get(HConstants.ZOOKEEPER_ZNODE_PARENT, HConstants.DEFAULT_ZOOKEEPER_ZNODE_PARENT)); HBaseSink sink = new HBaseSink(); Configurables.configure(sink, tmpContext); Assert.fail(); }
@SuppressWarnings("unchecked") @Override public void configure(Context context) { if (!HBaseVersionCheck.hasVersionLessThan2(logger)) { throw new ConfigurationException( "HBase major version number must be less than 2 for hbase-sink."); } tableName = context.getString(HBaseSinkConfigurationConstants.CONFIG_TABLE); String cf = context.getString( HBaseSinkConfigurationConstants.CONFIG_COLUMN_FAMILY); batchSize = context.getLong( HBaseSinkConfigurationConstants.CONFIG_BATCHSIZE, new Long(100)); serializerContext = new Context(); eventSerializerType = context.getString( HBaseSinkConfigurationConstants.CONFIG_SERIALIZER); Preconditions.checkNotNull(tableName, "Table name cannot be empty, please specify in configuration file"); Preconditions.checkNotNull(cf, "Column family cannot be empty, please specify in configuration file"); if (eventSerializerType == null || eventSerializerType.isEmpty()) { eventSerializerType = "org.apache.flume.sink.hbase.SimpleHbaseEventSerializer"; logger.info("No serializer defined, Will use default"); } serializerContext.putAll(context.getSubProperties( HBaseSinkConfigurationConstants.CONFIG_SERIALIZER_PREFIX)); columnFamily = cf.getBytes(Charsets.UTF_8); try { Class<? extends HbaseEventSerializer> clazz = (Class<? extends HbaseEventSerializer>) Class.forName(eventSerializerType); serializer = clazz.newInstance(); serializer.configure(serializerContext); } catch (Exception e) { logger.error("Could not instantiate event serializer.", e); Throwables.propagate(e); } kerberosKeytab = context.getString(HBaseSinkConfigurationConstants.CONFIG_KEYTAB); kerberosPrincipal = context.getString(HBaseSinkConfigurationConstants.CONFIG_PRINCIPAL); enableWal = context.getBoolean(HBaseSinkConfigurationConstants .CONFIG_ENABLE_WAL, HBaseSinkConfigurationConstants.DEFAULT_ENABLE_WAL); logger.info("The write to WAL option is set to: " + String.valueOf(enableWal)); if (!enableWal) { logger.warn("HBase Sink's enableWal configuration is set to false. All " + "writes to HBase will have WAL disabled, and any data in the " + "memstore of this region in the Region Server could be lost!"); } batchIncrements = context.getBoolean( HBaseSinkConfigurationConstants.CONFIG_COALESCE_INCREMENTS, HBaseSinkConfigurationConstants.DEFAULT_COALESCE_INCREMENTS); if (batchIncrements) { logger.info("Increment coalescing is enabled. Increments will be " + "buffered."); refGetFamilyMap = reflectLookupGetFamilyMap(); } String zkQuorum = context.getString(HBaseSinkConfigurationConstants .ZK_QUORUM); Integer port = null; if (zkQuorum != null && !zkQuorum.isEmpty()) { StringBuilder zkBuilder = new StringBuilder(); logger.info("Using ZK Quorum: " + zkQuorum); String[] zkHosts = zkQuorum.split(","); int length = zkHosts.length; for (int i = 0; i < length; i++) { String[] zkHostAndPort = zkHosts[i].split(":"); zkBuilder.append(zkHostAndPort[0].trim()); if (i != length - 1) { zkBuilder.append(","); } else { zkQuorum = zkBuilder.toString(); } if (zkHostAndPort[1] == null) { throw new FlumeException("Expected client port for the ZK node!"); } if (port == null) { port = Integer.parseInt(zkHostAndPort[1].trim()); } else if (!port.equals(Integer.parseInt(zkHostAndPort[1].trim()))) { throw new FlumeException("All Zookeeper nodes in the quorum must " + "use the same client port."); } } if (port == null) { port = HConstants.DEFAULT_ZOOKEPER_CLIENT_PORT; } this.config.set(HConstants.ZOOKEEPER_QUORUM, zkQuorum); this.config.setInt(HConstants.ZOOKEEPER_CLIENT_PORT, port); } String hbaseZnode = context.getString( HBaseSinkConfigurationConstants.ZK_ZNODE_PARENT); if (hbaseZnode != null && !hbaseZnode.isEmpty()) { this.config.set(HConstants.ZOOKEEPER_ZNODE_PARENT, hbaseZnode); } sinkCounter = new SinkCounter(this.getName()); }
HBaseSink extends AbstractSink implements Configurable, BatchSizeSupported { @SuppressWarnings("unchecked") @Override public void configure(Context context) { if (!HBaseVersionCheck.hasVersionLessThan2(logger)) { throw new ConfigurationException( "HBase major version number must be less than 2 for hbase-sink."); } tableName = context.getString(HBaseSinkConfigurationConstants.CONFIG_TABLE); String cf = context.getString( HBaseSinkConfigurationConstants.CONFIG_COLUMN_FAMILY); batchSize = context.getLong( HBaseSinkConfigurationConstants.CONFIG_BATCHSIZE, new Long(100)); serializerContext = new Context(); eventSerializerType = context.getString( HBaseSinkConfigurationConstants.CONFIG_SERIALIZER); Preconditions.checkNotNull(tableName, "Table name cannot be empty, please specify in configuration file"); Preconditions.checkNotNull(cf, "Column family cannot be empty, please specify in configuration file"); if (eventSerializerType == null || eventSerializerType.isEmpty()) { eventSerializerType = "org.apache.flume.sink.hbase.SimpleHbaseEventSerializer"; logger.info("No serializer defined, Will use default"); } serializerContext.putAll(context.getSubProperties( HBaseSinkConfigurationConstants.CONFIG_SERIALIZER_PREFIX)); columnFamily = cf.getBytes(Charsets.UTF_8); try { Class<? extends HbaseEventSerializer> clazz = (Class<? extends HbaseEventSerializer>) Class.forName(eventSerializerType); serializer = clazz.newInstance(); serializer.configure(serializerContext); } catch (Exception e) { logger.error("Could not instantiate event serializer.", e); Throwables.propagate(e); } kerberosKeytab = context.getString(HBaseSinkConfigurationConstants.CONFIG_KEYTAB); kerberosPrincipal = context.getString(HBaseSinkConfigurationConstants.CONFIG_PRINCIPAL); enableWal = context.getBoolean(HBaseSinkConfigurationConstants .CONFIG_ENABLE_WAL, HBaseSinkConfigurationConstants.DEFAULT_ENABLE_WAL); logger.info("The write to WAL option is set to: " + String.valueOf(enableWal)); if (!enableWal) { logger.warn("HBase Sink's enableWal configuration is set to false. All " + "writes to HBase will have WAL disabled, and any data in the " + "memstore of this region in the Region Server could be lost!"); } batchIncrements = context.getBoolean( HBaseSinkConfigurationConstants.CONFIG_COALESCE_INCREMENTS, HBaseSinkConfigurationConstants.DEFAULT_COALESCE_INCREMENTS); if (batchIncrements) { logger.info("Increment coalescing is enabled. Increments will be " + "buffered."); refGetFamilyMap = reflectLookupGetFamilyMap(); } String zkQuorum = context.getString(HBaseSinkConfigurationConstants .ZK_QUORUM); Integer port = null; if (zkQuorum != null && !zkQuorum.isEmpty()) { StringBuilder zkBuilder = new StringBuilder(); logger.info("Using ZK Quorum: " + zkQuorum); String[] zkHosts = zkQuorum.split(","); int length = zkHosts.length; for (int i = 0; i < length; i++) { String[] zkHostAndPort = zkHosts[i].split(":"); zkBuilder.append(zkHostAndPort[0].trim()); if (i != length - 1) { zkBuilder.append(","); } else { zkQuorum = zkBuilder.toString(); } if (zkHostAndPort[1] == null) { throw new FlumeException("Expected client port for the ZK node!"); } if (port == null) { port = Integer.parseInt(zkHostAndPort[1].trim()); } else if (!port.equals(Integer.parseInt(zkHostAndPort[1].trim()))) { throw new FlumeException("All Zookeeper nodes in the quorum must " + "use the same client port."); } } if (port == null) { port = HConstants.DEFAULT_ZOOKEPER_CLIENT_PORT; } this.config.set(HConstants.ZOOKEEPER_QUORUM, zkQuorum); this.config.setInt(HConstants.ZOOKEEPER_CLIENT_PORT, port); } String hbaseZnode = context.getString( HBaseSinkConfigurationConstants.ZK_ZNODE_PARENT); if (hbaseZnode != null && !hbaseZnode.isEmpty()) { this.config.set(HConstants.ZOOKEEPER_ZNODE_PARENT, hbaseZnode); } sinkCounter = new SinkCounter(this.getName()); } }
HBaseSink extends AbstractSink implements Configurable, BatchSizeSupported { @SuppressWarnings("unchecked") @Override public void configure(Context context) { if (!HBaseVersionCheck.hasVersionLessThan2(logger)) { throw new ConfigurationException( "HBase major version number must be less than 2 for hbase-sink."); } tableName = context.getString(HBaseSinkConfigurationConstants.CONFIG_TABLE); String cf = context.getString( HBaseSinkConfigurationConstants.CONFIG_COLUMN_FAMILY); batchSize = context.getLong( HBaseSinkConfigurationConstants.CONFIG_BATCHSIZE, new Long(100)); serializerContext = new Context(); eventSerializerType = context.getString( HBaseSinkConfigurationConstants.CONFIG_SERIALIZER); Preconditions.checkNotNull(tableName, "Table name cannot be empty, please specify in configuration file"); Preconditions.checkNotNull(cf, "Column family cannot be empty, please specify in configuration file"); if (eventSerializerType == null || eventSerializerType.isEmpty()) { eventSerializerType = "org.apache.flume.sink.hbase.SimpleHbaseEventSerializer"; logger.info("No serializer defined, Will use default"); } serializerContext.putAll(context.getSubProperties( HBaseSinkConfigurationConstants.CONFIG_SERIALIZER_PREFIX)); columnFamily = cf.getBytes(Charsets.UTF_8); try { Class<? extends HbaseEventSerializer> clazz = (Class<? extends HbaseEventSerializer>) Class.forName(eventSerializerType); serializer = clazz.newInstance(); serializer.configure(serializerContext); } catch (Exception e) { logger.error("Could not instantiate event serializer.", e); Throwables.propagate(e); } kerberosKeytab = context.getString(HBaseSinkConfigurationConstants.CONFIG_KEYTAB); kerberosPrincipal = context.getString(HBaseSinkConfigurationConstants.CONFIG_PRINCIPAL); enableWal = context.getBoolean(HBaseSinkConfigurationConstants .CONFIG_ENABLE_WAL, HBaseSinkConfigurationConstants.DEFAULT_ENABLE_WAL); logger.info("The write to WAL option is set to: " + String.valueOf(enableWal)); if (!enableWal) { logger.warn("HBase Sink's enableWal configuration is set to false. All " + "writes to HBase will have WAL disabled, and any data in the " + "memstore of this region in the Region Server could be lost!"); } batchIncrements = context.getBoolean( HBaseSinkConfigurationConstants.CONFIG_COALESCE_INCREMENTS, HBaseSinkConfigurationConstants.DEFAULT_COALESCE_INCREMENTS); if (batchIncrements) { logger.info("Increment coalescing is enabled. Increments will be " + "buffered."); refGetFamilyMap = reflectLookupGetFamilyMap(); } String zkQuorum = context.getString(HBaseSinkConfigurationConstants .ZK_QUORUM); Integer port = null; if (zkQuorum != null && !zkQuorum.isEmpty()) { StringBuilder zkBuilder = new StringBuilder(); logger.info("Using ZK Quorum: " + zkQuorum); String[] zkHosts = zkQuorum.split(","); int length = zkHosts.length; for (int i = 0; i < length; i++) { String[] zkHostAndPort = zkHosts[i].split(":"); zkBuilder.append(zkHostAndPort[0].trim()); if (i != length - 1) { zkBuilder.append(","); } else { zkQuorum = zkBuilder.toString(); } if (zkHostAndPort[1] == null) { throw new FlumeException("Expected client port for the ZK node!"); } if (port == null) { port = Integer.parseInt(zkHostAndPort[1].trim()); } else if (!port.equals(Integer.parseInt(zkHostAndPort[1].trim()))) { throw new FlumeException("All Zookeeper nodes in the quorum must " + "use the same client port."); } } if (port == null) { port = HConstants.DEFAULT_ZOOKEPER_CLIENT_PORT; } this.config.set(HConstants.ZOOKEEPER_QUORUM, zkQuorum); this.config.setInt(HConstants.ZOOKEEPER_CLIENT_PORT, port); } String hbaseZnode = context.getString( HBaseSinkConfigurationConstants.ZK_ZNODE_PARENT); if (hbaseZnode != null && !hbaseZnode.isEmpty()) { this.config.set(HConstants.ZOOKEEPER_ZNODE_PARENT, hbaseZnode); } sinkCounter = new SinkCounter(this.getName()); } HBaseSink();  HBaseSink(Configuration conf); @VisibleForTesting @InterfaceAudience.Private  HBaseSink(Configuration conf, DebugIncrementsCallback cb); }
HBaseSink extends AbstractSink implements Configurable, BatchSizeSupported { @SuppressWarnings("unchecked") @Override public void configure(Context context) { if (!HBaseVersionCheck.hasVersionLessThan2(logger)) { throw new ConfigurationException( "HBase major version number must be less than 2 for hbase-sink."); } tableName = context.getString(HBaseSinkConfigurationConstants.CONFIG_TABLE); String cf = context.getString( HBaseSinkConfigurationConstants.CONFIG_COLUMN_FAMILY); batchSize = context.getLong( HBaseSinkConfigurationConstants.CONFIG_BATCHSIZE, new Long(100)); serializerContext = new Context(); eventSerializerType = context.getString( HBaseSinkConfigurationConstants.CONFIG_SERIALIZER); Preconditions.checkNotNull(tableName, "Table name cannot be empty, please specify in configuration file"); Preconditions.checkNotNull(cf, "Column family cannot be empty, please specify in configuration file"); if (eventSerializerType == null || eventSerializerType.isEmpty()) { eventSerializerType = "org.apache.flume.sink.hbase.SimpleHbaseEventSerializer"; logger.info("No serializer defined, Will use default"); } serializerContext.putAll(context.getSubProperties( HBaseSinkConfigurationConstants.CONFIG_SERIALIZER_PREFIX)); columnFamily = cf.getBytes(Charsets.UTF_8); try { Class<? extends HbaseEventSerializer> clazz = (Class<? extends HbaseEventSerializer>) Class.forName(eventSerializerType); serializer = clazz.newInstance(); serializer.configure(serializerContext); } catch (Exception e) { logger.error("Could not instantiate event serializer.", e); Throwables.propagate(e); } kerberosKeytab = context.getString(HBaseSinkConfigurationConstants.CONFIG_KEYTAB); kerberosPrincipal = context.getString(HBaseSinkConfigurationConstants.CONFIG_PRINCIPAL); enableWal = context.getBoolean(HBaseSinkConfigurationConstants .CONFIG_ENABLE_WAL, HBaseSinkConfigurationConstants.DEFAULT_ENABLE_WAL); logger.info("The write to WAL option is set to: " + String.valueOf(enableWal)); if (!enableWal) { logger.warn("HBase Sink's enableWal configuration is set to false. All " + "writes to HBase will have WAL disabled, and any data in the " + "memstore of this region in the Region Server could be lost!"); } batchIncrements = context.getBoolean( HBaseSinkConfigurationConstants.CONFIG_COALESCE_INCREMENTS, HBaseSinkConfigurationConstants.DEFAULT_COALESCE_INCREMENTS); if (batchIncrements) { logger.info("Increment coalescing is enabled. Increments will be " + "buffered."); refGetFamilyMap = reflectLookupGetFamilyMap(); } String zkQuorum = context.getString(HBaseSinkConfigurationConstants .ZK_QUORUM); Integer port = null; if (zkQuorum != null && !zkQuorum.isEmpty()) { StringBuilder zkBuilder = new StringBuilder(); logger.info("Using ZK Quorum: " + zkQuorum); String[] zkHosts = zkQuorum.split(","); int length = zkHosts.length; for (int i = 0; i < length; i++) { String[] zkHostAndPort = zkHosts[i].split(":"); zkBuilder.append(zkHostAndPort[0].trim()); if (i != length - 1) { zkBuilder.append(","); } else { zkQuorum = zkBuilder.toString(); } if (zkHostAndPort[1] == null) { throw new FlumeException("Expected client port for the ZK node!"); } if (port == null) { port = Integer.parseInt(zkHostAndPort[1].trim()); } else if (!port.equals(Integer.parseInt(zkHostAndPort[1].trim()))) { throw new FlumeException("All Zookeeper nodes in the quorum must " + "use the same client port."); } } if (port == null) { port = HConstants.DEFAULT_ZOOKEPER_CLIENT_PORT; } this.config.set(HConstants.ZOOKEEPER_QUORUM, zkQuorum); this.config.setInt(HConstants.ZOOKEEPER_CLIENT_PORT, port); } String hbaseZnode = context.getString( HBaseSinkConfigurationConstants.ZK_ZNODE_PARENT); if (hbaseZnode != null && !hbaseZnode.isEmpty()) { this.config.set(HConstants.ZOOKEEPER_ZNODE_PARENT, hbaseZnode); } sinkCounter = new SinkCounter(this.getName()); } HBaseSink();  HBaseSink(Configuration conf); @VisibleForTesting @InterfaceAudience.Private  HBaseSink(Configuration conf, DebugIncrementsCallback cb); @Override void start(); @Override void stop(); @SuppressWarnings("unchecked") @Override void configure(Context context); Configuration getConfig(); @Override Status process(); @Override long getBatchSize(); }
HBaseSink extends AbstractSink implements Configurable, BatchSizeSupported { @SuppressWarnings("unchecked") @Override public void configure(Context context) { if (!HBaseVersionCheck.hasVersionLessThan2(logger)) { throw new ConfigurationException( "HBase major version number must be less than 2 for hbase-sink."); } tableName = context.getString(HBaseSinkConfigurationConstants.CONFIG_TABLE); String cf = context.getString( HBaseSinkConfigurationConstants.CONFIG_COLUMN_FAMILY); batchSize = context.getLong( HBaseSinkConfigurationConstants.CONFIG_BATCHSIZE, new Long(100)); serializerContext = new Context(); eventSerializerType = context.getString( HBaseSinkConfigurationConstants.CONFIG_SERIALIZER); Preconditions.checkNotNull(tableName, "Table name cannot be empty, please specify in configuration file"); Preconditions.checkNotNull(cf, "Column family cannot be empty, please specify in configuration file"); if (eventSerializerType == null || eventSerializerType.isEmpty()) { eventSerializerType = "org.apache.flume.sink.hbase.SimpleHbaseEventSerializer"; logger.info("No serializer defined, Will use default"); } serializerContext.putAll(context.getSubProperties( HBaseSinkConfigurationConstants.CONFIG_SERIALIZER_PREFIX)); columnFamily = cf.getBytes(Charsets.UTF_8); try { Class<? extends HbaseEventSerializer> clazz = (Class<? extends HbaseEventSerializer>) Class.forName(eventSerializerType); serializer = clazz.newInstance(); serializer.configure(serializerContext); } catch (Exception e) { logger.error("Could not instantiate event serializer.", e); Throwables.propagate(e); } kerberosKeytab = context.getString(HBaseSinkConfigurationConstants.CONFIG_KEYTAB); kerberosPrincipal = context.getString(HBaseSinkConfigurationConstants.CONFIG_PRINCIPAL); enableWal = context.getBoolean(HBaseSinkConfigurationConstants .CONFIG_ENABLE_WAL, HBaseSinkConfigurationConstants.DEFAULT_ENABLE_WAL); logger.info("The write to WAL option is set to: " + String.valueOf(enableWal)); if (!enableWal) { logger.warn("HBase Sink's enableWal configuration is set to false. All " + "writes to HBase will have WAL disabled, and any data in the " + "memstore of this region in the Region Server could be lost!"); } batchIncrements = context.getBoolean( HBaseSinkConfigurationConstants.CONFIG_COALESCE_INCREMENTS, HBaseSinkConfigurationConstants.DEFAULT_COALESCE_INCREMENTS); if (batchIncrements) { logger.info("Increment coalescing is enabled. Increments will be " + "buffered."); refGetFamilyMap = reflectLookupGetFamilyMap(); } String zkQuorum = context.getString(HBaseSinkConfigurationConstants .ZK_QUORUM); Integer port = null; if (zkQuorum != null && !zkQuorum.isEmpty()) { StringBuilder zkBuilder = new StringBuilder(); logger.info("Using ZK Quorum: " + zkQuorum); String[] zkHosts = zkQuorum.split(","); int length = zkHosts.length; for (int i = 0; i < length; i++) { String[] zkHostAndPort = zkHosts[i].split(":"); zkBuilder.append(zkHostAndPort[0].trim()); if (i != length - 1) { zkBuilder.append(","); } else { zkQuorum = zkBuilder.toString(); } if (zkHostAndPort[1] == null) { throw new FlumeException("Expected client port for the ZK node!"); } if (port == null) { port = Integer.parseInt(zkHostAndPort[1].trim()); } else if (!port.equals(Integer.parseInt(zkHostAndPort[1].trim()))) { throw new FlumeException("All Zookeeper nodes in the quorum must " + "use the same client port."); } } if (port == null) { port = HConstants.DEFAULT_ZOOKEPER_CLIENT_PORT; } this.config.set(HConstants.ZOOKEEPER_QUORUM, zkQuorum); this.config.setInt(HConstants.ZOOKEEPER_CLIENT_PORT, port); } String hbaseZnode = context.getString( HBaseSinkConfigurationConstants.ZK_ZNODE_PARENT); if (hbaseZnode != null && !hbaseZnode.isEmpty()) { this.config.set(HConstants.ZOOKEEPER_ZNODE_PARENT, hbaseZnode); } sinkCounter = new SinkCounter(this.getName()); } HBaseSink();  HBaseSink(Configuration conf); @VisibleForTesting @InterfaceAudience.Private  HBaseSink(Configuration conf, DebugIncrementsCallback cb); @Override void start(); @Override void stop(); @SuppressWarnings("unchecked") @Override void configure(Context context); Configuration getConfig(); @Override Status process(); @Override long getBatchSize();  }