BDREHDFSEventSink extends AbstractSink implements Configurable { @Override @SuppressWarnings("squid:MethodCyclomaticComplexity") public void configure(Context context) { this.context = context; filePath = Preconditions.checkNotNull( context.getString("hdfs.path"), "hdfs.path is required"); fileName = context.getString("hdfs.filePrefix", DEFAULT_FILE_NAME); this.suffix = context.getString("hdfs.fileSuffix", DEFAULT_SUFFIX); inUsePrefix = context.getString("hdfs.inUsePrefix", DEFAULT_IN_USE_PREFIX); inUseSuffix = context.getString("hdfs.inUseSuffix", DEFAULT_IN_USE_SUFFIX); String tzName = context.getString("hdfs.timeZone"); timeZone = tzName == null ? null : TimeZone.getTimeZone(tzName); rollInterval = context.getLong("hdfs.rollInterval", DEFAULT_ROLL_INTERVAL); rollSize = context.getLong("hdfs.rollSize", DEFAULT_ROLL_SIZE); rollCount = context.getLong("hdfs.rollCount", DEFAULT_ROLL_COUNT); batchSize = context.getLong("hdfs.batchSize", DEFAULT_BATCH_SIZE); idleTimeout = context.getInteger("hdfs.idleTimeout", 0); String codecName = context.getString("hdfs.codeC"); fileType = context.getString("hdfs.fileType", DEFAULT_FILE_TYPE); maxOpenFiles = context.getInteger("hdfs.maxOpenFiles", DEFAULT_MAX_OPEN_FILES); callTimeout = context.getLong("hdfs.callTimeout", DEFAULT_CALL_TIMEOUT); threadsPoolSize = context.getInteger("hdfs.threadsPoolSize", DEFAULT_THREAD_POOL_SIZE); rollTimerPoolSize = context.getInteger("hdfs.rollTimerPoolSize", DEFAULT_ROLL_TIMER_POOL_SIZE); String kerbConfPrincipal = context.getString("hdfs.kerberosPrincipal"); String kerbKeytab = context.getString("hdfs.kerberosKeytab"); String proxyUser = context.getString("hdfs.proxyUser"); tryCount = context.getInteger("hdfs.closeTries", DEFAULT_TRY_COUNT); processId = context.getString("hdfs.processId"); if(tryCount <= 0) { LOGGER.warn("Retry count value : " + tryCount + " is not " + "valid. The sink will try to close the file until the file " + "is eventually closed."); tryCount = DEFAULT_TRY_COUNT; } retryInterval = context.getLong("hdfs.retryInterval", DEFAULT_RETRY_INTERVAL); if(retryInterval <= 0) { LOGGER.warn("Retry Interval value: " + retryInterval + " is not " + "valid. If the first close of a file fails, " + "it may remain open and will not be renamed."); tryCount = 1; } 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.DATA_STREAM_TYPE) && 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.COMP_STREAM_TYPE)) { Preconditions.checkNotNull(codeC, "It's essential to set compress codec" + " when fileType is: " + fileType); } this.privExecutor = FlumeAuthenticationUtil.getAuthenticator( kerbConfPrincipal, kerbKeytab).proxyAs(proxyUser); needRounding = context.getBoolean("hdfs.round", false); if(needRounding) { String unit = context.getString("hdfs.roundUnit", "second"); if ("hour".equalsIgnoreCase(unit)) { this.roundUnit = Calendar.HOUR_OF_DAY; } else if ("minute".equalsIgnoreCase(unit)) { this.roundUnit = Calendar.MINUTE; } else if ("second".equalsIgnoreCase(unit)){ this.roundUnit = Calendar.SECOND; } else { LOGGER.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()); } } BDREHDFSEventSink();  BDREHDFSEventSink(HDFSWriterFactory writerFactory); @Override @SuppressWarnings("squid:MethodCyclomaticComplexity") void configure(Context context); @Override @SuppressWarnings("squid:MethodCyclomaticComplexity") Status process(); @Override void stop(); @Override void start(); @Override String toString();  }
@Test @Ignore 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( "Keytab is not a readable file")); } finally { conf.set(CommonConfigurationKeys.HADOOP_SECURITY_AUTHENTICATION, "simple"); UserGroupInformation.setConfiguration(conf); } }