FlinkEnvManager { public FlinkEnvManager(String configPath, String jobName, String inTable, String outTable) throws IOException { this.inTable = inTable; this.outTable = outTable; this.jobName = jobName; readConfig(configPath); } FlinkEnvManager(String configPath, String jobName,
                           String inTable, String outTable);  FlinkEnvManager(String configPath); ExecutionEnvironment getExecutionEnvironment(); DataSet<Tuple2<Key,Value>> getDataFromAccumulo(); Connector getConnector(); DataSet<Tuple2<Key,Value>> getDataFromAccumulo(ExecutionEnvironment env); HadoopOutputFormat getHadoopOF();  }
@Test public void testFlinkEnvManager() throws Exception{ tmpSettingsDir.create(); File settings = tmpSettingsDir.newFile("settings"); Connector conn = amc.getConnector(); FileOutputStream fos = new FileOutputStream(settings, false); BufferedWriter br = new BufferedWriter(new OutputStreamWriter(fos)); String parameters = "accumulo.instance=" + conn.getInstance().getInstanceName() + "\n"+ "accumulo.user=" + conn.whoami() +"\n"+ "accumulo.password=password\n"+ "accumulo.zookeeper=" + conn.getInstance().getZooKeepers(); System.out.println(parameters); br.write(parameters); br.flush(); br.close(); fos.flush(); fos.close(); FlinkEnvManager fem = new FlinkEnvManager(settings.getAbsolutePath(),"testJob", TableIdentifier.RAW_TWITTER_DATA.get(), TableIdentifier.TERM_INDEX.get()); assertNotNull(fem.getDataFromAccumulo()); assertNotNull(fem.getExecutionEnvironment()); assertNotNull(fem.getHadoopOF()); }