core/src/test/java/org/apache/oozie/service/TestHadoopAccessorService.java - oozie - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package org.apache.oozie.service;

 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.ipc.RemoteException;
 import org.apache.hadoop.mapreduce.MRJobConfig;
 import org.apache.hadoop.security.authorize.AuthorizationException;
 import org.apache.hadoop.yarn.api.records.LocalResource;
 import org.apache.hadoop.yarn.api.records.LocalResourceType;
 import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
 import org.apache.hadoop.yarn.client.api.YarnClient;
 import org.apache.hadoop.yarn.util.ConverterUtils;
 import org.apache.oozie.test.XFsTestCase;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.JobClient;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.io.Text;
 import org.apache.oozie.util.IOUtils;

 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.net.URI;

 import org.apache.hadoop.conf.Configuration;
 import org.apache.oozie.ErrorCode;
 import org.apache.oozie.util.XConfiguration;
 import org.junit.Assert;

 public class TestHadoopAccessorService extends XFsTestCase {

     protected void setUp() throws Exception {
         super.setUp();
         new File(getTestCaseConfDir(), "hadoop-confx").mkdir();
         File actConfXDir = new File(getTestCaseConfDir(), "action-confx");
         actConfXDir.mkdir();
         new File(actConfXDir, "action").mkdir();
         // Create the default action dir
         new File(actConfXDir, "default").mkdir();
         InputStream is = Thread.currentThread().getContextClassLoader().getResourceAsStream("test-hadoop-config.xml");
         OutputStream os = new FileOutputStream(new File(getTestCaseConfDir() + "/hadoop-confx", "core-site.xml"));
         IOUtils.copyStream(is, os);
         is = Thread.currentThread().getContextClassLoader().getResourceAsStream("test-action-config.xml");
         os = new FileOutputStream(new File(actConfXDir, "action.xml"));
         IOUtils.copyStream(is, os);

         is = Thread.currentThread().getContextClassLoader().getResourceAsStream("test-default-config.xml");
         os = new FileOutputStream(new File(actConfXDir, "default.xml"));
         IOUtils.copyStream(is, os);

         is = Thread.currentThread().getContextClassLoader().getResourceAsStream("test-action-config-1.xml");
         os = new FileOutputStream(new File(actConfXDir + "/action", "a-conf.xml"));
         IOUtils.copyStream(is, os);

         is = Thread.currentThread().getContextClassLoader().getResourceAsStream("test-action-config-2.xml");
         os = new FileOutputStream(new File(actConfXDir + "/action", "b-conf.xml"));
         IOUtils.copyStream(is, os);

         is = Thread.currentThread().getContextClassLoader().getResourceAsStream("test-action-config-3.xml");
         os = new FileOutputStream(new File(actConfXDir + "/action", "c-conf-3.xml"));
         IOUtils.copyStream(is, os);

         is = Thread.currentThread().getContextClassLoader().getResourceAsStream("test-default-config-1.xml");
         os = new FileOutputStream(new File(actConfXDir + "/default", "z-conf.xml"));
         IOUtils.copyStream(is, os);

         is = Thread.currentThread().getContextClassLoader().getResourceAsStream("test-custom-log4j.properties");
         os = new FileOutputStream(new File(actConfXDir + "/action", "test-custom-log4j.properties"));
         IOUtils.copyStream(is, os);

         setSystemProperty("oozie.service.HadoopAccessorService.hadoop.configurations",
                           "*=hadoop-conf,jt=hadoop-confx");
         setSystemProperty("oozie.service.HadoopAccessorService.action.configurations",
                           "*=hadoop-conf,jt=action-confx");
         if (System.getProperty("oozie.test.hadoop.security", "simple").equals("kerberos")) {
             setSystemProperty("oozie.service.HadoopAccessorService.kerberos.enabled", "true");
             setSystemProperty("oozie.service.HadoopAccessorService.keytab.file", getKeytabFile());
             setSystemProperty("oozie.service.HadoopAccessorService.kerberos.principal", getOoziePrincipal());
         }
         Services services = new Services();
         services.init();
     }

     protected void tearDown() throws Exception {
         Services.get().destroy();
         super.tearDown();
     }

     public void testService() throws Exception {
         Services services = Services.get();
         HadoopAccessorService has = services.get(HadoopAccessorService.class);
         assertNotNull(has);
         assertNotNull(has.createConfiguration("*"));
         assertNotNull(has.createConfiguration("jt"));
         assertEquals("bar", has.createConfiguration("jt").get("foo"));
         assertNotNull(has.createActionDefaultConf("*", "action"));
         assertNotNull(has.createActionDefaultConf("jt", "action"));
         assertNotNull(has.createActionDefaultConf("jt", "actionx"));
         assertNotNull(has.createActionDefaultConf("jtx", "action"));
         assertNull(has.createActionDefaultConf("*", "action").get("action.foo"));
     }

     public void testActionConfigurations() throws Exception {
         Services services = Services.get();
         HadoopAccessorService has = services.get(HadoopAccessorService.class);
         assertNotNull(has);
         XConfiguration conf = has.createActionDefaultConf("jt", "action");
         assertNotNull(conf);

         // Check that the param only in default.xml is still present
         assertEquals("default.bar", conf.get("default.foo"));

         // And a property that is present in one of the conf files in default dir
         assertEquals("default.bus", conf.get("default.car"));

         // Check that a default param is overridden by one of the action config files
         assertEquals("action.bar", conf.get("action.foo"));

         // Check that params from <action-dir>/action/conf files is still present
         assertEquals("action.car", conf.get("action.boo"));
         assertEquals("action.carcar", conf.get("oozie.launcher.action.booboo"));

         /*
             Check precedence - Order of precedence - 0 is the lowest.   Parameters in files of
             lower precedence will be overridden by redefinitions in higher precedence files.

             0 - All files in defaultdir/*.xml (sorted by lexical name)
                Files with names lexically lower have lesser precedence than the following ones.
             1 - default.xml
             2 - All files in actiondir/*.xml (sort by lexical name)
                Files with names lexically lower have lesser precedence than the following ones
             3 - action.xml
          */
         assertEquals("100", conf.get("action.testprop"));
         assertEquals("1", conf.get("default.testprop"));
         assertEquals("org.apache.log4j.ConsoleAppender", conf.get("log4j.appender.oozie"));
         assertEquals("NONE, null", conf.get("log4j.logger.a"));
     }

     public void testCreateJobClient() throws Exception {
         HadoopAccessorService has = Services.get().get(HadoopAccessorService.class);
         Configuration conf = has.createConfiguration(getJobTrackerUri());

         JobClient jc = has.createJobClient(getTestUser(), conf);
         assertNotNull(jc);
         jc.getAllJobs();

         JobConf conf2 = new JobConf(false);
         conf2.set("mapred.job.tracker", getJobTrackerUri());
         try {
             has.createJobClient(getTestUser(), conf2);
             fail("Should have thrown exception because Configuration not created by HadoopAccessorService");
         }
         catch (HadoopAccessorException ex) {
             assertEquals(ErrorCode.E0903, ex.getErrorCode());
         }
     }

     public void testCreateYarnClient() throws Exception {
         HadoopAccessorService has = Services.get().get(HadoopAccessorService.class);
         Configuration conf = has.createConfiguration(getJobTrackerUri());

         YarnClient yc = has.createYarnClient(getTestUser(), conf);
         assertNotNull(yc);
         yc.getApplications();

         try {
             yc = has.createYarnClient("invalid-user", conf);
             assertNotNull(yc);
             yc.getApplications();
             fail("Should have thrown exception because not allowed to impersonate 'invalid-user'");
         }
         catch (AuthorizationException ex) {
         }

         JobConf conf2 = new JobConf(false);
         conf2.set("yarn.resourcemanager.address", getJobTrackerUri());
         try {
             has.createYarnClient(getTestUser(), conf2);
             fail("Should have thrown exception because Configuration not created by HadoopAccessorService");
         }
         catch (HadoopAccessorException ex) {
             assertEquals(ErrorCode.E0903, ex.getErrorCode());
         }
     }

     public void testCreateFileSystem() throws Exception {
         HadoopAccessorService has = Services.get().get(HadoopAccessorService.class);
         Configuration conf = has.createConfiguration(getJobTrackerUri());

         FileSystem fs = has.createFileSystem(getTestUser(), new URI(getNameNodeUri()), conf);
         assertNotNull(fs);
         fs.exists(new Path(getNameNodeUri(), "/foo"));

         try {
             fs = has.createFileSystem("invalid-user", new URI(getNameNodeUri()), conf);
             assertNotNull(fs);
             fs.exists(new Path(getNameNodeUri(), "/foo"));
             fail("Should have thrown exception because not allowed to impersonate 'invalid-user'");
         }
         catch (RemoteException ex) {
             assertEquals(AuthorizationException.class.getName(), ex.getClassName());
         }

         JobConf conf2 = new JobConf(false);
         conf2.set("fs.default.name", getNameNodeUri());
         try {
             has.createFileSystem(getTestUser(), new URI(getNameNodeUri()), conf2);
             fail("Should have thrown exception because Configuration not created by HadoopAccessorService");
         }
         catch (HadoopAccessorException ex) {
             assertEquals(ErrorCode.E0903, ex.getErrorCode());
         }
     }

     public void testCheckSupportedFilesystem() throws Exception {
         Configuration hConf = Services.get().getConf();

         // Only allow hdfs and foo schemes
         HadoopAccessorService has = new HadoopAccessorService();
         hConf.set("oozie.service.HadoopAccessorService.supported.filesystems", "hdfs,foo");
         has.init(hConf);
         has.checkSupportedFilesystem(new URI("hdfs://localhost:1234/blah"));
         has.checkSupportedFilesystem(new URI("foo://localhost:1234/blah"));
         try {
             has.checkSupportedFilesystem(new URI("file://localhost:1234/blah"));
             fail("Should have thrown an exception because 'file' scheme isn't allowed");
         }
         catch (HadoopAccessorException hae) {
             assertEquals(ErrorCode.E0904, hae.getErrorCode());
         }
         // giving no scheme should skip the check
         has.checkSupportedFilesystem(new URI("/blah"));

         // allow all schemes
         has = new HadoopAccessorService();
         hConf.set("oozie.service.HadoopAccessorService.supported.filesystems", "*");
         has.init(hConf);
         has.checkSupportedFilesystem(new URI("hdfs://localhost:1234/blah"));
         has.checkSupportedFilesystem(new URI("foo://localhost:1234/blah"));
         has.checkSupportedFilesystem(new URI("file://localhost:1234/blah"));
         // giving no scheme should skip the check
         has.checkSupportedFilesystem(new URI("/blah"));
     }

     public void testValidateJobTracker() throws Exception {
         HadoopAccessorService has = new HadoopAccessorService();
         Configuration conf = new Configuration(false);
         conf.set("oozie.service.HadoopAccessorService.jobTracker.whitelist", " ");
         has.init(conf);
         has.validateJobTracker("foo");
         has.validateJobTracker("bar");
         has.validateJobTracker("blah");
         conf.set("oozie.service.HadoopAccessorService.jobTracker.whitelist", "foo,bar");
         has.init(conf);
         has.validateJobTracker("foo");
         has.validateJobTracker("bar");
         try {
             has.validateJobTracker("blah");
             fail("Should have gotten an exception");
         } catch (HadoopAccessorException hae) {
             assertEquals(ErrorCode.E0900, hae.getErrorCode());
             // We have to check for either of these because Java 7 and 8 have a different order
             String s1 = "E0900: JobTracker [blah] not allowed, not in Oozie's whitelist. Allowed values are: [foo, bar]";
             String s2 = "E0900: JobTracker [blah] not allowed, not in Oozie's whitelist. Allowed values are: [bar, foo]";
             assertTrue("expected:<" + s1 + "> or <" + s2 + "> but was:<" + hae.getMessage() + ">",
                     s1.equals(hae.getMessage()) || s2.equals(hae.getMessage()));
         }
         has.destroy();
     }

     public void testValidateNameNode() throws Exception {
         HadoopAccessorService has = new HadoopAccessorService();
         Configuration conf = new Configuration(false);
         conf.set("oozie.service.HadoopAccessorService.nameNode.whitelist", " ");
         has.init(conf);
         has.validateNameNode("foo");
         has.validateNameNode("bar");
         has.validateNameNode("blah");
         conf.set("oozie.service.HadoopAccessorService.nameNode.whitelist", "foo,bar");
         has.init(conf);
         has.validateNameNode("foo");
         has.validateNameNode("bar");
         try {
             has.validateNameNode("blah");
             fail("Should have gotten an exception");
         } catch (HadoopAccessorException hae) {
             assertEquals(ErrorCode.E0901, hae.getErrorCode());
             // We have to check for either of these because Java 7 and 8 have a different order
             String s1 = "E0901: NameNode [blah] not allowed, not in Oozie's whitelist. Allowed values are: [foo, bar]";
             String s2 = "E0901: NameNode [blah] not allowed, not in Oozie's whitelist. Allowed values are: [bar, foo]";
             assertTrue("expected:<" + s1 + "> or <" + s2 + "> but was:<" + hae.getMessage() + ">",
                     s1.equals(hae.getMessage()) || s2.equals(hae.getMessage()));
         }
         has.destroy();
     }

     public void testCreateLocalResourceForConfigurationFile() throws Exception {
         HadoopAccessorService has = Services.get().get(HadoopAccessorService.class);
         String filename = "foo.xml";
         Configuration conf = has.createConfiguration(getNameNodeUri());
         conf.set("foo", "bar");
         LocalResource lRes = has.createLocalResourceForConfigurationFile(filename, getTestUser(), conf, getFileSystem().getUri(),
                 getFsTestCaseDir());
         assertNotNull(lRes);
         assertEquals(LocalResourceType.FILE, lRes.getType());
         assertEquals(LocalResourceVisibility.APPLICATION, lRes.getVisibility());
         Path resPath = ConverterUtils.getPathFromYarnURL(lRes.getResource());
         assertEquals(new Path(getFsTestCaseDir(), "foo.xml"), resPath);
         Configuration conf2 = new Configuration(false);
         conf2.addResource(getFileSystem().open(resPath));
         assertEquals("bar", conf2.get("foo"));
     }

     public void testCreateS3WithByteBufferProperties() throws Exception {
         FileSystem fs = createFileSystemWithCustomProperties(
                 "s3a://somebucket/somedirectory/somefile.txt",
                 String.format(HadoopAccessorService.FS_PROP_PATTERN, "s3a"),
                 "fs.s3a.fast.upload.buffer=bytebuffer,fs.s3a.impl.disable.cache=true");
         assertEquals("Expected peroperty [bytebuffer] is not set.", "bytebuffer", fs.getConf().get("fs.s3a.fast.upload.buffer"));
     }

     public void testCreateS3WithDefaultProperties() throws Exception {
         FileSystem fs = createFileSystemWithCustomProperties("s3a://somebucket/somedirectory/somefile.txt", null, null);
         assertNotSame("Unxpected peroperty [bytebuffer] is set.", "bytebuffer", fs.getConf().get("fs.s3a.fast.upload.buffer"));
     }

     public void testCreateHdfsWithoutProperties() throws Exception {
         createFileSystemWithCustomProperties("hdfs://localhost:1234/somedirectory/somefile.txt", null, null);
     }

     public void testCreateFileWithoutProperties() throws Exception {
         createFileSystemWithCustomProperties("file://somebucket/somedirectory/somefile.txt", null, null);
     }

     public void testCreateWithoutSchemeWithoutProperties() throws Exception {
         createFileSystemWithCustomProperties("/somebucket/somedirectory/somefile.txt", null, null);
     }

     public void testCreateHdfsWithInvalidProperties() throws Exception {
         FileSystem fs = createFileSystemWithCustomProperties("hdfs://localhost:1234/somedirectory/somefile.txt",
                 String.format(HadoopAccessorService.FS_PROP_PATTERN, "hdfs"),
                 "fs.hdfs.custom.property1=value1,fs.hdfs.custom.property2");
         Assert.assertEquals("value1", fs.getConf().get("fs.hdfs.custom.property1"));
         Assert.assertEquals(null, fs.getConf().get("fs.hdfs.custom.property2"));
     }

     public void testCreateHdfsWithEqualSignInValuePropertiy() throws Exception {
         FileSystem fs = createFileSystemWithCustomProperties("hdfs://localhost:1234/somedirectory/somefile.txt",
                 String.format(HadoopAccessorService.FS_PROP_PATTERN, "hdfs"),
                 "fs.hdfs.custom.property1=value1=value2");
         Assert.assertEquals("value1=value2", fs.getConf().get("fs.hdfs.custom.property1"));
         Assert.assertEquals(null, fs.getConf().get("value1"));
         Assert.assertEquals(null, fs.getConf().get("value2"));
     }

     public void testCreateHdfsWithCommaSeparatedValues() throws Exception {
         FileSystem fs = createFileSystemWithCustomProperties("hdfs://localhost:1234/somedirectory/somefile.txt",
                 String.format(HadoopAccessorService.FS_PROP_PATTERN, "hdfs"),
                 "fs.hdfs.custom.property1=value1,value2");
         Assert.assertEquals("value1", fs.getConf().get("fs.hdfs.custom.property1"));
         Assert.assertEquals(null, fs.getConf().get("value2"));
     }

     public void testIfNoCustomFsConfigProvidedBaseConfigRemainsTheSame() throws Exception {
         HadoopAccessorService has = new HadoopAccessorService();
         Configuration base = new XConfiguration();
         base.set("foo.bar", "baz");
         Configuration result = has.extendWithFileSystemSpecificPropertiesIfAny(new URI("hdfs://localhost:1234/"), base);
         assertEquals("The two configuration object shall be the same", base, result);
         assertEquals("Key foo.bar shall be present in result configuration", "baz", result.get("foo.bar"));
         assertSame("The two configuration object shall be the same", base, result);
     }

     public void testIfMRLimitsIsInitialized() throws IOException, ServiceException {
         InputStream is = Thread.currentThread().getContextClassLoader().getResourceAsStream("test-mapred-site.xml");
         OutputStream os = new FileOutputStream(new File(getTestCaseConfDir() + "/hadoop-confx", "mapred-site.xml"));
         IOUtils.copyStream(is, os);
         HadoopAccessorService has = new HadoopAccessorService();
         has.init(Services.get());
         Assert.assertEquals("Limits class shall not use default value for number of counters.",
                 500,
                 has.createConfiguration("jt").getInt(MRJobConfig.COUNTERS_MAX_KEY, 120));
     }

     public FileSystem createFileSystemWithCustomProperties(
             final String uri, final String fsKey, final String commaSeparatedKeyValues) throws Exception {
         final HadoopAccessorService has = new HadoopAccessorService();
         if (fsKey != null && commaSeparatedKeyValues != null) {
             ConfigurationService.set(fsKey, commaSeparatedKeyValues);
         }
         has.init(new Configuration(false));
         final Configuration conf = has.createConfiguration(null);
         setS3CredentialProperties(conf);
         final FileSystem fs = has.createFileSystem("user", new URI(uri), conf);
         has.destroy();
         return fs;
     }

     private void setS3CredentialProperties(final Configuration conf) {
         conf.set("fs.s3a.access.key", "someAccessKey");
         conf.set("fs.s3a.secret.key", "someSecretKey");
     }
 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package org.apache.oozie.service;

	import org.apache.hadoop.fs.Path;
	import org.apache.hadoop.ipc.RemoteException;
	import org.apache.hadoop.mapreduce.MRJobConfig;
	import org.apache.hadoop.security.authorize.AuthorizationException;
	import org.apache.hadoop.yarn.api.records.LocalResource;
	import org.apache.hadoop.yarn.api.records.LocalResourceType;
	import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
	import org.apache.hadoop.yarn.client.api.YarnClient;
	import org.apache.hadoop.yarn.util.ConverterUtils;
	import org.apache.oozie.test.XFsTestCase;
	import org.apache.hadoop.mapred.JobConf;
	import org.apache.hadoop.mapred.JobClient;
	import org.apache.hadoop.fs.FileSystem;
	import org.apache.hadoop.io.Text;
	import org.apache.oozie.util.IOUtils;

	import java.io.File;
	import java.io.FileOutputStream;
	import java.io.IOException;
	import java.io.InputStream;
	import java.io.OutputStream;
	import java.net.URI;

	import org.apache.hadoop.conf.Configuration;
	import org.apache.oozie.ErrorCode;
	import org.apache.oozie.util.XConfiguration;
	import org.junit.Assert;

	public class TestHadoopAccessorService extends XFsTestCase {

	protected void setUp() throws Exception {
	super.setUp();
	new File(getTestCaseConfDir(), "hadoop-confx").mkdir();
	File actConfXDir = new File(getTestCaseConfDir(), "action-confx");
	actConfXDir.mkdir();
	new File(actConfXDir, "action").mkdir();
	// Create the default action dir
	new File(actConfXDir, "default").mkdir();
	InputStream is = Thread.currentThread().getContextClassLoader().getResourceAsStream("test-hadoop-config.xml");
	OutputStream os = new FileOutputStream(new File(getTestCaseConfDir() + "/hadoop-confx", "core-site.xml"));
	IOUtils.copyStream(is, os);
	is = Thread.currentThread().getContextClassLoader().getResourceAsStream("test-action-config.xml");
	os = new FileOutputStream(new File(actConfXDir, "action.xml"));
	IOUtils.copyStream(is, os);

	is = Thread.currentThread().getContextClassLoader().getResourceAsStream("test-default-config.xml");
	os = new FileOutputStream(new File(actConfXDir, "default.xml"));
	IOUtils.copyStream(is, os);

	is = Thread.currentThread().getContextClassLoader().getResourceAsStream("test-action-config-1.xml");
	os = new FileOutputStream(new File(actConfXDir + "/action", "a-conf.xml"));
	IOUtils.copyStream(is, os);

	is = Thread.currentThread().getContextClassLoader().getResourceAsStream("test-action-config-2.xml");
	os = new FileOutputStream(new File(actConfXDir + "/action", "b-conf.xml"));
	IOUtils.copyStream(is, os);

	is = Thread.currentThread().getContextClassLoader().getResourceAsStream("test-action-config-3.xml");
	os = new FileOutputStream(new File(actConfXDir + "/action", "c-conf-3.xml"));
	IOUtils.copyStream(is, os);

	is = Thread.currentThread().getContextClassLoader().getResourceAsStream("test-default-config-1.xml");
	os = new FileOutputStream(new File(actConfXDir + "/default", "z-conf.xml"));
	IOUtils.copyStream(is, os);

	is = Thread.currentThread().getContextClassLoader().getResourceAsStream("test-custom-log4j.properties");
	os = new FileOutputStream(new File(actConfXDir + "/action", "test-custom-log4j.properties"));
	IOUtils.copyStream(is, os);

	setSystemProperty("oozie.service.HadoopAccessorService.hadoop.configurations",
	"*=hadoop-conf,jt=hadoop-confx");
	setSystemProperty("oozie.service.HadoopAccessorService.action.configurations",
	"*=hadoop-conf,jt=action-confx");
	if (System.getProperty("oozie.test.hadoop.security", "simple").equals("kerberos")) {
	setSystemProperty("oozie.service.HadoopAccessorService.kerberos.enabled", "true");
	setSystemProperty("oozie.service.HadoopAccessorService.keytab.file", getKeytabFile());
	setSystemProperty("oozie.service.HadoopAccessorService.kerberos.principal", getOoziePrincipal());
	}
	Services services = new Services();
	services.init();
	}

	protected void tearDown() throws Exception {
	Services.get().destroy();
	super.tearDown();
	}

	public void testService() throws Exception {
	Services services = Services.get();
	HadoopAccessorService has = services.get(HadoopAccessorService.class);
	assertNotNull(has);
	assertNotNull(has.createConfiguration("*"));
	assertNotNull(has.createConfiguration("jt"));
	assertEquals("bar", has.createConfiguration("jt").get("foo"));
	assertNotNull(has.createActionDefaultConf("*", "action"));
	assertNotNull(has.createActionDefaultConf("jt", "action"));
	assertNotNull(has.createActionDefaultConf("jt", "actionx"));
	assertNotNull(has.createActionDefaultConf("jtx", "action"));
	assertNull(has.createActionDefaultConf("*", "action").get("action.foo"));
	}

	public void testActionConfigurations() throws Exception {
	Services services = Services.get();
	HadoopAccessorService has = services.get(HadoopAccessorService.class);
	assertNotNull(has);
	XConfiguration conf = has.createActionDefaultConf("jt", "action");
	assertNotNull(conf);

	// Check that the param only in default.xml is still present
	assertEquals("default.bar", conf.get("default.foo"));

	// And a property that is present in one of the conf files in default dir
	assertEquals("default.bus", conf.get("default.car"));

	// Check that a default param is overridden by one of the action config files
	assertEquals("action.bar", conf.get("action.foo"));

	// Check that params from <action-dir>/action/conf files is still present
	assertEquals("action.car", conf.get("action.boo"));
	assertEquals("action.carcar", conf.get("oozie.launcher.action.booboo"));

	/*
	Check precedence - Order of precedence - 0 is the lowest. Parameters in files of
	lower precedence will be overridden by redefinitions in higher precedence files.

	0 - All files in defaultdir/*.xml (sorted by lexical name)
	Files with names lexically lower have lesser precedence than the following ones.
	1 - default.xml
	2 - All files in actiondir/*.xml (sort by lexical name)
	Files with names lexically lower have lesser precedence than the following ones
	3 - action.xml
	*/
	assertEquals("100", conf.get("action.testprop"));
	assertEquals("1", conf.get("default.testprop"));
	assertEquals("org.apache.log4j.ConsoleAppender", conf.get("log4j.appender.oozie"));
	assertEquals("NONE, null", conf.get("log4j.logger.a"));
	}

	public void testCreateJobClient() throws Exception {
	HadoopAccessorService has = Services.get().get(HadoopAccessorService.class);
	Configuration conf = has.createConfiguration(getJobTrackerUri());

	JobClient jc = has.createJobClient(getTestUser(), conf);
	assertNotNull(jc);
	jc.getAllJobs();

	JobConf conf2 = new JobConf(false);
	conf2.set("mapred.job.tracker", getJobTrackerUri());
	try {
	has.createJobClient(getTestUser(), conf2);
	fail("Should have thrown exception because Configuration not created by HadoopAccessorService");
	}
	catch (HadoopAccessorException ex) {
	assertEquals(ErrorCode.E0903, ex.getErrorCode());
	}
	}

	public void testCreateYarnClient() throws Exception {
	HadoopAccessorService has = Services.get().get(HadoopAccessorService.class);
	Configuration conf = has.createConfiguration(getJobTrackerUri());

	YarnClient yc = has.createYarnClient(getTestUser(), conf);
	assertNotNull(yc);
	yc.getApplications();

	try {
	yc = has.createYarnClient("invalid-user", conf);
	assertNotNull(yc);
	yc.getApplications();
	fail("Should have thrown exception because not allowed to impersonate 'invalid-user'");
	}
	catch (AuthorizationException ex) {
	}

	JobConf conf2 = new JobConf(false);
	conf2.set("yarn.resourcemanager.address", getJobTrackerUri());
	try {
	has.createYarnClient(getTestUser(), conf2);
	fail("Should have thrown exception because Configuration not created by HadoopAccessorService");
	}
	catch (HadoopAccessorException ex) {
	assertEquals(ErrorCode.E0903, ex.getErrorCode());
	}
	}

	public void testCreateFileSystem() throws Exception {
	HadoopAccessorService has = Services.get().get(HadoopAccessorService.class);
	Configuration conf = has.createConfiguration(getJobTrackerUri());

	FileSystem fs = has.createFileSystem(getTestUser(), new URI(getNameNodeUri()), conf);
	assertNotNull(fs);
	fs.exists(new Path(getNameNodeUri(), "/foo"));

	try {
	fs = has.createFileSystem("invalid-user", new URI(getNameNodeUri()), conf);
	assertNotNull(fs);
	fs.exists(new Path(getNameNodeUri(), "/foo"));
	fail("Should have thrown exception because not allowed to impersonate 'invalid-user'");
	}
	catch (RemoteException ex) {
	assertEquals(AuthorizationException.class.getName(), ex.getClassName());
	}

	JobConf conf2 = new JobConf(false);
	conf2.set("fs.default.name", getNameNodeUri());
	try {
	has.createFileSystem(getTestUser(), new URI(getNameNodeUri()), conf2);
	fail("Should have thrown exception because Configuration not created by HadoopAccessorService");
	}
	catch (HadoopAccessorException ex) {
	assertEquals(ErrorCode.E0903, ex.getErrorCode());
	}
	}

	public void testCheckSupportedFilesystem() throws Exception {
	Configuration hConf = Services.get().getConf();

	// Only allow hdfs and foo schemes
	HadoopAccessorService has = new HadoopAccessorService();
	hConf.set("oozie.service.HadoopAccessorService.supported.filesystems", "hdfs,foo");
	has.init(hConf);
	has.checkSupportedFilesystem(new URI("hdfs://localhost:1234/blah"));
	has.checkSupportedFilesystem(new URI("foo://localhost:1234/blah"));
	try {
	has.checkSupportedFilesystem(new URI("file://localhost:1234/blah"));
	fail("Should have thrown an exception because 'file' scheme isn't allowed");
	}
	catch (HadoopAccessorException hae) {
	assertEquals(ErrorCode.E0904, hae.getErrorCode());
	}
	// giving no scheme should skip the check
	has.checkSupportedFilesystem(new URI("/blah"));

	// allow all schemes
	has = new HadoopAccessorService();
	hConf.set("oozie.service.HadoopAccessorService.supported.filesystems", "*");
	has.init(hConf);
	has.checkSupportedFilesystem(new URI("hdfs://localhost:1234/blah"));
	has.checkSupportedFilesystem(new URI("foo://localhost:1234/blah"));
	has.checkSupportedFilesystem(new URI("file://localhost:1234/blah"));
	// giving no scheme should skip the check
	has.checkSupportedFilesystem(new URI("/blah"));
	}

	public void testValidateJobTracker() throws Exception {
	HadoopAccessorService has = new HadoopAccessorService();
	Configuration conf = new Configuration(false);
	conf.set("oozie.service.HadoopAccessorService.jobTracker.whitelist", " ");
	has.init(conf);
	has.validateJobTracker("foo");
	has.validateJobTracker("bar");
	has.validateJobTracker("blah");
	conf.set("oozie.service.HadoopAccessorService.jobTracker.whitelist", "foo,bar");
	has.init(conf);
	has.validateJobTracker("foo");
	has.validateJobTracker("bar");
	try {
	has.validateJobTracker("blah");
	fail("Should have gotten an exception");
	} catch (HadoopAccessorException hae) {
	assertEquals(ErrorCode.E0900, hae.getErrorCode());
	// We have to check for either of these because Java 7 and 8 have a different order
	String s1 = "E0900: JobTracker [blah] not allowed, not in Oozie's whitelist. Allowed values are: [foo, bar]";
	String s2 = "E0900: JobTracker [blah] not allowed, not in Oozie's whitelist. Allowed values are: [bar, foo]";
	assertTrue("expected:<" + s1 + "> or <" + s2 + "> but was:<" + hae.getMessage() + ">",
	s1.equals(hae.getMessage()) \|\| s2.equals(hae.getMessage()));
	}
	has.destroy();
	}

	public void testValidateNameNode() throws Exception {
	HadoopAccessorService has = new HadoopAccessorService();
	Configuration conf = new Configuration(false);
	conf.set("oozie.service.HadoopAccessorService.nameNode.whitelist", " ");
	has.init(conf);
	has.validateNameNode("foo");
	has.validateNameNode("bar");
	has.validateNameNode("blah");
	conf.set("oozie.service.HadoopAccessorService.nameNode.whitelist", "foo,bar");
	has.init(conf);
	has.validateNameNode("foo");
	has.validateNameNode("bar");
	try {
	has.validateNameNode("blah");
	fail("Should have gotten an exception");
	} catch (HadoopAccessorException hae) {
	assertEquals(ErrorCode.E0901, hae.getErrorCode());
	// We have to check for either of these because Java 7 and 8 have a different order
	String s1 = "E0901: NameNode [blah] not allowed, not in Oozie's whitelist. Allowed values are: [foo, bar]";
	String s2 = "E0901: NameNode [blah] not allowed, not in Oozie's whitelist. Allowed values are: [bar, foo]";
	assertTrue("expected:<" + s1 + "> or <" + s2 + "> but was:<" + hae.getMessage() + ">",
	s1.equals(hae.getMessage()) \|\| s2.equals(hae.getMessage()));
	}
	has.destroy();
	}

	public void testCreateLocalResourceForConfigurationFile() throws Exception {
	HadoopAccessorService has = Services.get().get(HadoopAccessorService.class);
	String filename = "foo.xml";
	Configuration conf = has.createConfiguration(getNameNodeUri());
	conf.set("foo", "bar");
	LocalResource lRes = has.createLocalResourceForConfigurationFile(filename, getTestUser(), conf, getFileSystem().getUri(),
	getFsTestCaseDir());
	assertNotNull(lRes);
	assertEquals(LocalResourceType.FILE, lRes.getType());
	assertEquals(LocalResourceVisibility.APPLICATION, lRes.getVisibility());
	Path resPath = ConverterUtils.getPathFromYarnURL(lRes.getResource());
	assertEquals(new Path(getFsTestCaseDir(), "foo.xml"), resPath);
	Configuration conf2 = new Configuration(false);
	conf2.addResource(getFileSystem().open(resPath));
	assertEquals("bar", conf2.get("foo"));
	}

	public void testCreateS3WithByteBufferProperties() throws Exception {
	FileSystem fs = createFileSystemWithCustomProperties(
	"s3a://somebucket/somedirectory/somefile.txt",
	String.format(HadoopAccessorService.FS_PROP_PATTERN, "s3a"),
	"fs.s3a.fast.upload.buffer=bytebuffer,fs.s3a.impl.disable.cache=true");
	assertEquals("Expected peroperty [bytebuffer] is not set.", "bytebuffer", fs.getConf().get("fs.s3a.fast.upload.buffer"));
	}

	public void testCreateS3WithDefaultProperties() throws Exception {
	FileSystem fs = createFileSystemWithCustomProperties("s3a://somebucket/somedirectory/somefile.txt", null, null);
	assertNotSame("Unxpected peroperty [bytebuffer] is set.", "bytebuffer", fs.getConf().get("fs.s3a.fast.upload.buffer"));
	}

	public void testCreateHdfsWithoutProperties() throws Exception {
	createFileSystemWithCustomProperties("hdfs://localhost:1234/somedirectory/somefile.txt", null, null);
	}

	public void testCreateFileWithoutProperties() throws Exception {
	createFileSystemWithCustomProperties("file://somebucket/somedirectory/somefile.txt", null, null);
	}

	public void testCreateWithoutSchemeWithoutProperties() throws Exception {
	createFileSystemWithCustomProperties("/somebucket/somedirectory/somefile.txt", null, null);
	}

	public void testCreateHdfsWithInvalidProperties() throws Exception {
	FileSystem fs = createFileSystemWithCustomProperties("hdfs://localhost:1234/somedirectory/somefile.txt",
	String.format(HadoopAccessorService.FS_PROP_PATTERN, "hdfs"),
	"fs.hdfs.custom.property1=value1,fs.hdfs.custom.property2");
	Assert.assertEquals("value1", fs.getConf().get("fs.hdfs.custom.property1"));
	Assert.assertEquals(null, fs.getConf().get("fs.hdfs.custom.property2"));
	}

	public void testCreateHdfsWithEqualSignInValuePropertiy() throws Exception {
	FileSystem fs = createFileSystemWithCustomProperties("hdfs://localhost:1234/somedirectory/somefile.txt",
	String.format(HadoopAccessorService.FS_PROP_PATTERN, "hdfs"),
	"fs.hdfs.custom.property1=value1=value2");
	Assert.assertEquals("value1=value2", fs.getConf().get("fs.hdfs.custom.property1"));
	Assert.assertEquals(null, fs.getConf().get("value1"));
	Assert.assertEquals(null, fs.getConf().get("value2"));
	}

	public void testCreateHdfsWithCommaSeparatedValues() throws Exception {
	FileSystem fs = createFileSystemWithCustomProperties("hdfs://localhost:1234/somedirectory/somefile.txt",
	String.format(HadoopAccessorService.FS_PROP_PATTERN, "hdfs"),
	"fs.hdfs.custom.property1=value1,value2");
	Assert.assertEquals("value1", fs.getConf().get("fs.hdfs.custom.property1"));
	Assert.assertEquals(null, fs.getConf().get("value2"));
	}

	public void testIfNoCustomFsConfigProvidedBaseConfigRemainsTheSame() throws Exception {
	HadoopAccessorService has = new HadoopAccessorService();
	Configuration base = new XConfiguration();
	base.set("foo.bar", "baz");
	Configuration result = has.extendWithFileSystemSpecificPropertiesIfAny(new URI("hdfs://localhost:1234/"), base);
	assertEquals("The two configuration object shall be the same", base, result);
	assertEquals("Key foo.bar shall be present in result configuration", "baz", result.get("foo.bar"));
	assertSame("The two configuration object shall be the same", base, result);
	}

	public void testIfMRLimitsIsInitialized() throws IOException, ServiceException {
	InputStream is = Thread.currentThread().getContextClassLoader().getResourceAsStream("test-mapred-site.xml");
	OutputStream os = new FileOutputStream(new File(getTestCaseConfDir() + "/hadoop-confx", "mapred-site.xml"));
	IOUtils.copyStream(is, os);
	HadoopAccessorService has = new HadoopAccessorService();
	has.init(Services.get());
	Assert.assertEquals("Limits class shall not use default value for number of counters.",
	500,
	has.createConfiguration("jt").getInt(MRJobConfig.COUNTERS_MAX_KEY, 120));
	}

	public FileSystem createFileSystemWithCustomProperties(
	final String uri, final String fsKey, final String commaSeparatedKeyValues) throws Exception {
	final HadoopAccessorService has = new HadoopAccessorService();
	if (fsKey != null && commaSeparatedKeyValues != null) {
	ConfigurationService.set(fsKey, commaSeparatedKeyValues);
	}
	has.init(new Configuration(false));
	final Configuration conf = has.createConfiguration(null);
	setS3CredentialProperties(conf);
	final FileSystem fs = has.createFileSystem("user", new URI(uri), conf);
	has.destroy();
	return fs;
	}

	private void setS3CredentialProperties(final Configuration conf) {
	conf.set("fs.s3a.access.key", "someAccessKey");
	conf.set("fs.s3a.secret.key", "someSecretKey");
	}
	}