@Test public void testKerbFileAccess() throws InterruptedException, LifecycleException, EventDeliveryException, IOException { LOG.debug("Starting testKerbFileAccess() ..."); final String fileName = "FlumeData"; final long rollCount = 5; final long batchSize = 2; String newPath = testPath + "/singleBucket"; String kerbConfPrincipal = "user1/localhost@EXAMPLE.COM"; String kerbKeytab = "/usr/lib/flume/nonexistkeytabfile"; Configuration conf = new Configuration(); conf.set(CommonConfigurationKeys.HADOOP_SECURITY_AUTHENTICATION, "kerberos"); UserGroupInformation.setConfiguration(conf); Context context = new Context(); context.put("hdfs.path", newPath); context.put("hdfs.filePrefix", fileName); context.put("hdfs.rollCount", String.valueOf(rollCount)); context.put("hdfs.batchSize", String.valueOf(batchSize)); context.put("hdfs.kerberosPrincipal", kerbConfPrincipal); context.put("hdfs.kerberosKeytab", kerbKeytab); try { Configurables.configure(sink, context); Assert.fail("no exception thrown"); } catch (IllegalArgumentException expected) { Assert.assertTrue(expected.getMessage().contains( "is nonexistent or can't read.")); } finally { conf.set(CommonConfigurationKeys.HADOOP_SECURITY_AUTHENTICATION, "simple"); UserGroupInformation.setConfiguration(conf); } }
@Override public void configure(Context context) { this.context = context; filePath = Preconditions.checkNotNull( context.getString("hdfs.path"), "hdfs.path is required"); fileName = context.getString("hdfs.filePrefix", defaultFileName); this.suffix = context.getString("hdfs.fileSuffix", defaultSuffix); inUsePrefix = context.getString("hdfs.inUsePrefix", defaultInUsePrefix); inUseSuffix = context.getString("hdfs.inUseSuffix", defaultInUseSuffix); String tzName = context.getString("hdfs.timeZone"); timeZone = tzName == null ? null : TimeZone.getTimeZone(tzName); rollInterval = context.getLong("hdfs.rollInterval", defaultRollInterval); rollSize = context.getLong("hdfs.rollSize", defaultRollSize); rollCount = context.getLong("hdfs.rollCount", defaultRollCount); batchSize = context.getLong("hdfs.batchSize", defaultBatchSize); idleTimeout = context.getInteger("hdfs.idleTimeout", 0); String codecName = context.getString("hdfs.codeC"); fileType = context.getString("hdfs.fileType", defaultFileType); maxOpenFiles = context.getInteger("hdfs.maxOpenFiles", defaultMaxOpenFiles); callTimeout = context.getLong("hdfs.callTimeout", defaultCallTimeout); threadsPoolSize = context.getInteger("hdfs.threadsPoolSize", defaultThreadPoolSize); rollTimerPoolSize = context.getInteger("hdfs.rollTimerPoolSize", defaultRollTimerPoolSize); kerbConfPrincipal = context.getString("hdfs.kerberosPrincipal", ""); kerbKeytab = context.getString("hdfs.kerberosKeytab", ""); proxyUserName = context.getString("hdfs.proxyUser", ""); Preconditions.checkArgument(batchSize > 0, "batchSize must be greater than 0"); if (codecName == null) { codeC = null; compType = CompressionType.NONE; } else { codeC = getCodec(codecName); compType = CompressionType.BLOCK; } if(fileType.equalsIgnoreCase(HDFSWriterFactory.DataStreamType) && codecName != null) { throw new IllegalArgumentException("fileType: " + fileType + " which does NOT support compressed output. Please don't set codeC" + " or change the fileType if compressed output is desired."); } if(fileType.equalsIgnoreCase(HDFSWriterFactory.CompStreamType)) { Preconditions.checkNotNull(codeC, "It's essential to set compress codec" + " when fileType is: " + fileType); } if (!authenticate()) { LOG.error("Failed to authenticate!"); } needRounding = context.getBoolean("hdfs.round", false); if(needRounding) { String unit = context.getString("hdfs.roundUnit", "second"); if (unit.equalsIgnoreCase("hour")) { this.roundUnit = Calendar.HOUR_OF_DAY; } else if (unit.equalsIgnoreCase("minute")) { this.roundUnit = Calendar.MINUTE; } else if (unit.equalsIgnoreCase("second")){ this.roundUnit = Calendar.SECOND; } else { LOG.warn("Rounding unit is not valid, please set one of" + "minute, hour, or second. Rounding will be disabled"); needRounding = false; } this.roundValue = context.getInteger("hdfs.roundValue", 1); if(roundUnit == Calendar.SECOND || roundUnit == Calendar.MINUTE){ Preconditions.checkArgument(roundValue > 0 && roundValue <= 60, "Round value" + "must be > 0 and <= 60"); } else if (roundUnit == Calendar.HOUR_OF_DAY){ Preconditions.checkArgument(roundValue > 0 && roundValue <= 24, "Round value" + "must be > 0 and <= 24"); } } this.useLocalTime = context.getBoolean("hdfs.useLocalTimeStamp", false); if(useLocalTime) { clock = new SystemClock(); } if (sinkCounter == null) { sinkCounter = new SinkCounter(getName()); } }
HDFSEventSink extends AbstractSink implements Configurable { @Override public void configure(Context context) { this.context = context; filePath = Preconditions.checkNotNull( context.getString("hdfs.path"), "hdfs.path is required"); fileName = context.getString("hdfs.filePrefix", defaultFileName); this.suffix = context.getString("hdfs.fileSuffix", defaultSuffix); inUsePrefix = context.getString("hdfs.inUsePrefix", defaultInUsePrefix); inUseSuffix = context.getString("hdfs.inUseSuffix", defaultInUseSuffix); String tzName = context.getString("hdfs.timeZone"); timeZone = tzName == null ? null : TimeZone.getTimeZone(tzName); rollInterval = context.getLong("hdfs.rollInterval", defaultRollInterval); rollSize = context.getLong("hdfs.rollSize", defaultRollSize); rollCount = context.getLong("hdfs.rollCount", defaultRollCount); batchSize = context.getLong("hdfs.batchSize", defaultBatchSize); idleTimeout = context.getInteger("hdfs.idleTimeout", 0); String codecName = context.getString("hdfs.codeC"); fileType = context.getString("hdfs.fileType", defaultFileType); maxOpenFiles = context.getInteger("hdfs.maxOpenFiles", defaultMaxOpenFiles); callTimeout = context.getLong("hdfs.callTimeout", defaultCallTimeout); threadsPoolSize = context.getInteger("hdfs.threadsPoolSize", defaultThreadPoolSize); rollTimerPoolSize = context.getInteger("hdfs.rollTimerPoolSize", defaultRollTimerPoolSize); kerbConfPrincipal = context.getString("hdfs.kerberosPrincipal", ""); kerbKeytab = context.getString("hdfs.kerberosKeytab", ""); proxyUserName = context.getString("hdfs.proxyUser", ""); Preconditions.checkArgument(batchSize > 0, "batchSize must be greater than 0"); if (codecName == null) { codeC = null; compType = CompressionType.NONE; } else { codeC = getCodec(codecName); compType = CompressionType.BLOCK; } if(fileType.equalsIgnoreCase(HDFSWriterFactory.DataStreamType) && codecName != null) { throw new IllegalArgumentException("fileType: " + fileType + " which does NOT support compressed output. Please don't set codeC" + " or change the fileType if compressed output is desired."); } if(fileType.equalsIgnoreCase(HDFSWriterFactory.CompStreamType)) { Preconditions.checkNotNull(codeC, "It's essential to set compress codec" + " when fileType is: " + fileType); } if (!authenticate()) { LOG.error("Failed to authenticate!"); } needRounding = context.getBoolean("hdfs.round", false); if(needRounding) { String unit = context.getString("hdfs.roundUnit", "second"); if (unit.equalsIgnoreCase("hour")) { this.roundUnit = Calendar.HOUR_OF_DAY; } else if (unit.equalsIgnoreCase("minute")) { this.roundUnit = Calendar.MINUTE; } else if (unit.equalsIgnoreCase("second")){ this.roundUnit = Calendar.SECOND; } else { LOG.warn("Rounding unit is not valid, please set one of" + "minute, hour, or second. Rounding will be disabled"); needRounding = false; } this.roundValue = context.getInteger("hdfs.roundValue", 1); if(roundUnit == Calendar.SECOND || roundUnit == Calendar.MINUTE){ Preconditions.checkArgument(roundValue > 0 && roundValue <= 60, "Round value" + "must be > 0 and <= 60"); } else if (roundUnit == Calendar.HOUR_OF_DAY){ Preconditions.checkArgument(roundValue > 0 && roundValue <= 24, "Round value" + "must be > 0 and <= 24"); } } this.useLocalTime = context.getBoolean("hdfs.useLocalTimeStamp", false); if(useLocalTime) { clock = new SystemClock(); } if (sinkCounter == null) { sinkCounter = new SinkCounter(getName()); } } }
HDFSEventSink extends AbstractSink implements Configurable { @Override public void configure(Context context) { this.context = context; filePath = Preconditions.checkNotNull( context.getString("hdfs.path"), "hdfs.path is required"); fileName = context.getString("hdfs.filePrefix", defaultFileName); this.suffix = context.getString("hdfs.fileSuffix", defaultSuffix); inUsePrefix = context.getString("hdfs.inUsePrefix", defaultInUsePrefix); inUseSuffix = context.getString("hdfs.inUseSuffix", defaultInUseSuffix); String tzName = context.getString("hdfs.timeZone"); timeZone = tzName == null ? null : TimeZone.getTimeZone(tzName); rollInterval = context.getLong("hdfs.rollInterval", defaultRollInterval); rollSize = context.getLong("hdfs.rollSize", defaultRollSize); rollCount = context.getLong("hdfs.rollCount", defaultRollCount); batchSize = context.getLong("hdfs.batchSize", defaultBatchSize); idleTimeout = context.getInteger("hdfs.idleTimeout", 0); String codecName = context.getString("hdfs.codeC"); fileType = context.getString("hdfs.fileType", defaultFileType); maxOpenFiles = context.getInteger("hdfs.maxOpenFiles", defaultMaxOpenFiles); callTimeout = context.getLong("hdfs.callTimeout", defaultCallTimeout); threadsPoolSize = context.getInteger("hdfs.threadsPoolSize", defaultThreadPoolSize); rollTimerPoolSize = context.getInteger("hdfs.rollTimerPoolSize", defaultRollTimerPoolSize); kerbConfPrincipal = context.getString("hdfs.kerberosPrincipal", ""); kerbKeytab = context.getString("hdfs.kerberosKeytab", ""); proxyUserName = context.getString("hdfs.proxyUser", ""); Preconditions.checkArgument(batchSize > 0, "batchSize must be greater than 0"); if (codecName == null) { codeC = null; compType = CompressionType.NONE; } else { codeC = getCodec(codecName); compType = CompressionType.BLOCK; } if(fileType.equalsIgnoreCase(HDFSWriterFactory.DataStreamType) && codecName != null) { throw new IllegalArgumentException("fileType: " + fileType + " which does NOT support compressed output. Please don't set codeC" + " or change the fileType if compressed output is desired."); } if(fileType.equalsIgnoreCase(HDFSWriterFactory.CompStreamType)) { Preconditions.checkNotNull(codeC, "It's essential to set compress codec" + " when fileType is: " + fileType); } if (!authenticate()) { LOG.error("Failed to authenticate!"); } needRounding = context.getBoolean("hdfs.round", false); if(needRounding) { String unit = context.getString("hdfs.roundUnit", "second"); if (unit.equalsIgnoreCase("hour")) { this.roundUnit = Calendar.HOUR_OF_DAY; } else if (unit.equalsIgnoreCase("minute")) { this.roundUnit = Calendar.MINUTE; } else if (unit.equalsIgnoreCase("second")){ this.roundUnit = Calendar.SECOND; } else { LOG.warn("Rounding unit is not valid, please set one of" + "minute, hour, or second. Rounding will be disabled"); needRounding = false; } this.roundValue = context.getInteger("hdfs.roundValue", 1); if(roundUnit == Calendar.SECOND || roundUnit == Calendar.MINUTE){ Preconditions.checkArgument(roundValue > 0 && roundValue <= 60, "Round value" + "must be > 0 and <= 60"); } else if (roundUnit == Calendar.HOUR_OF_DAY){ Preconditions.checkArgument(roundValue > 0 && roundValue <= 24, "Round value" + "must be > 0 and <= 24"); } } this.useLocalTime = context.getBoolean("hdfs.useLocalTimeStamp", false); if(useLocalTime) { clock = new SystemClock(); } if (sinkCounter == null) { sinkCounter = new SinkCounter(getName()); } } HDFSEventSink();  HDFSEventSink(HDFSWriterFactory writerFactory); }
HDFSEventSink extends AbstractSink implements Configurable { @Override public void configure(Context context) { this.context = context; filePath = Preconditions.checkNotNull( context.getString("hdfs.path"), "hdfs.path is required"); fileName = context.getString("hdfs.filePrefix", defaultFileName); this.suffix = context.getString("hdfs.fileSuffix", defaultSuffix); inUsePrefix = context.getString("hdfs.inUsePrefix", defaultInUsePrefix); inUseSuffix = context.getString("hdfs.inUseSuffix", defaultInUseSuffix); String tzName = context.getString("hdfs.timeZone"); timeZone = tzName == null ? null : TimeZone.getTimeZone(tzName); rollInterval = context.getLong("hdfs.rollInterval", defaultRollInterval); rollSize = context.getLong("hdfs.rollSize", defaultRollSize); rollCount = context.getLong("hdfs.rollCount", defaultRollCount); batchSize = context.getLong("hdfs.batchSize", defaultBatchSize); idleTimeout = context.getInteger("hdfs.idleTimeout", 0); String codecName = context.getString("hdfs.codeC"); fileType = context.getString("hdfs.fileType", defaultFileType); maxOpenFiles = context.getInteger("hdfs.maxOpenFiles", defaultMaxOpenFiles); callTimeout = context.getLong("hdfs.callTimeout", defaultCallTimeout); threadsPoolSize = context.getInteger("hdfs.threadsPoolSize", defaultThreadPoolSize); rollTimerPoolSize = context.getInteger("hdfs.rollTimerPoolSize", defaultRollTimerPoolSize); kerbConfPrincipal = context.getString("hdfs.kerberosPrincipal", ""); kerbKeytab = context.getString("hdfs.kerberosKeytab", ""); proxyUserName = context.getString("hdfs.proxyUser", ""); Preconditions.checkArgument(batchSize > 0, "batchSize must be greater than 0"); if (codecName == null) { codeC = null; compType = CompressionType.NONE; } else { codeC = getCodec(codecName); compType = CompressionType.BLOCK; } if(fileType.equalsIgnoreCase(HDFSWriterFactory.DataStreamType) && codecName != null) { throw new IllegalArgumentException("fileType: " + fileType + " which does NOT support compressed output. Please don't set codeC" + " or change the fileType if compressed output is desired."); } if(fileType.equalsIgnoreCase(HDFSWriterFactory.CompStreamType)) { Preconditions.checkNotNull(codeC, "It's essential to set compress codec" + " when fileType is: " + fileType); } if (!authenticate()) { LOG.error("Failed to authenticate!"); } needRounding = context.getBoolean("hdfs.round", false); if(needRounding) { String unit = context.getString("hdfs.roundUnit", "second"); if (unit.equalsIgnoreCase("hour")) { this.roundUnit = Calendar.HOUR_OF_DAY; } else if (unit.equalsIgnoreCase("minute")) { this.roundUnit = Calendar.MINUTE; } else if (unit.equalsIgnoreCase("second")){ this.roundUnit = Calendar.SECOND; } else { LOG.warn("Rounding unit is not valid, please set one of" + "minute, hour, or second. Rounding will be disabled"); needRounding = false; } this.roundValue = context.getInteger("hdfs.roundValue", 1); if(roundUnit == Calendar.SECOND || roundUnit == Calendar.MINUTE){ Preconditions.checkArgument(roundValue > 0 && roundValue <= 60, "Round value" + "must be > 0 and <= 60"); } else if (roundUnit == Calendar.HOUR_OF_DAY){ Preconditions.checkArgument(roundValue > 0 && roundValue <= 24, "Round value" + "must be > 0 and <= 24"); } } this.useLocalTime = context.getBoolean("hdfs.useLocalTimeStamp", false); if(useLocalTime) { clock = new SystemClock(); } if (sinkCounter == null) { sinkCounter = new SinkCounter(getName()); } } HDFSEventSink();  HDFSEventSink(HDFSWriterFactory writerFactory); @Override void configure(Context context); Status process(); @Override void stop(); @Override void start(); @Override String toString(); }
HDFSEventSink extends AbstractSink implements Configurable { @Override public void configure(Context context) { this.context = context; filePath = Preconditions.checkNotNull( context.getString("hdfs.path"), "hdfs.path is required"); fileName = context.getString("hdfs.filePrefix", defaultFileName); this.suffix = context.getString("hdfs.fileSuffix", defaultSuffix); inUsePrefix = context.getString("hdfs.inUsePrefix", defaultInUsePrefix); inUseSuffix = context.getString("hdfs.inUseSuffix", defaultInUseSuffix); String tzName = context.getString("hdfs.timeZone"); timeZone = tzName == null ? null : TimeZone.getTimeZone(tzName); rollInterval = context.getLong("hdfs.rollInterval", defaultRollInterval); rollSize = context.getLong("hdfs.rollSize", defaultRollSize); rollCount = context.getLong("hdfs.rollCount", defaultRollCount); batchSize = context.getLong("hdfs.batchSize", defaultBatchSize); idleTimeout = context.getInteger("hdfs.idleTimeout", 0); String codecName = context.getString("hdfs.codeC"); fileType = context.getString("hdfs.fileType", defaultFileType); maxOpenFiles = context.getInteger("hdfs.maxOpenFiles", defaultMaxOpenFiles); callTimeout = context.getLong("hdfs.callTimeout", defaultCallTimeout); threadsPoolSize = context.getInteger("hdfs.threadsPoolSize", defaultThreadPoolSize); rollTimerPoolSize = context.getInteger("hdfs.rollTimerPoolSize", defaultRollTimerPoolSize); kerbConfPrincipal = context.getString("hdfs.kerberosPrincipal", ""); kerbKeytab = context.getString("hdfs.kerberosKeytab", ""); proxyUserName = context.getString("hdfs.proxyUser", ""); Preconditions.checkArgument(batchSize > 0, "batchSize must be greater than 0"); if (codecName == null) { codeC = null; compType = CompressionType.NONE; } else { codeC = getCodec(codecName); compType = CompressionType.BLOCK; } if(fileType.equalsIgnoreCase(HDFSWriterFactory.DataStreamType) && codecName != null) { throw new IllegalArgumentException("fileType: " + fileType + " which does NOT support compressed output. Please don't set codeC" + " or change the fileType if compressed output is desired."); } if(fileType.equalsIgnoreCase(HDFSWriterFactory.CompStreamType)) { Preconditions.checkNotNull(codeC, "It's essential to set compress codec" + " when fileType is: " + fileType); } if (!authenticate()) { LOG.error("Failed to authenticate!"); } needRounding = context.getBoolean("hdfs.round", false); if(needRounding) { String unit = context.getString("hdfs.roundUnit", "second"); if (unit.equalsIgnoreCase("hour")) { this.roundUnit = Calendar.HOUR_OF_DAY; } else if (unit.equalsIgnoreCase("minute")) { this.roundUnit = Calendar.MINUTE; } else if (unit.equalsIgnoreCase("second")){ this.roundUnit = Calendar.SECOND; } else { LOG.warn("Rounding unit is not valid, please set one of" + "minute, hour, or second. Rounding will be disabled"); needRounding = false; } this.roundValue = context.getInteger("hdfs.roundValue", 1); if(roundUnit == Calendar.SECOND || roundUnit == Calendar.MINUTE){ Preconditions.checkArgument(roundValue > 0 && roundValue <= 60, "Round value" + "must be > 0 and <= 60"); } else if (roundUnit == Calendar.HOUR_OF_DAY){ Preconditions.checkArgument(roundValue > 0 && roundValue <= 24, "Round value" + "must be > 0 and <= 24"); } } this.useLocalTime = context.getBoolean("hdfs.useLocalTimeStamp", false); if(useLocalTime) { clock = new SystemClock(); } if (sinkCounter == null) { sinkCounter = new SinkCounter(getName()); } } HDFSEventSink();  HDFSEventSink(HDFSWriterFactory writerFactory); @Override void configure(Context context); Status process(); @Override void stop(); @Override void start(); @Override String toString();  }