gobblin/gobblin-cluster

@RestLiCollection(name = "joblauncher", namespace = "gobblin.rest")
public class JobLauncherResource extends CollectionResourceTemplate<String, JobLauncherInfo> {

    private static final Logger LOGGER = LoggerFactory.getLogger(JobLauncherResource.class);

    static HashMap<String, Properties> jobPropertyMap = new HashMap<String, Properties>();

    // export GOBBLIN_JOB_CONFIG_DIR=/home/vickey/development/gobblin/gobblin-dist/job-config-bpu/wikipedia.pull
    // http://localhost:8080/joblauncher?q=triggerjob&jobId=wikipedia&search=india&propfile=gobblin-standalone.properties
    @Finder("triggerjob")
    public List<JobLauncherInfo> search(@PagingContextParam PagingContextParam context,
                                        @QueryParam("jobId") String jobId,
                                        @QueryParam("propfile") String propertiesFile,
                                        @QueryParam("fbToken") @Optional String fbToken,
                                        @QueryParam("twToken") @Optional String twToken,
                                        @QueryParam("twTokenSecret") @Optional String twTokenSecret,
                                        @QueryParam("nvToken") @Optional String nvToken,
                                        @QueryParam("fkToken") @Optional String fkToken,
                                        @QueryParam("fkTokenSecret") @Optional String fkTokenSecret,
                                        @QueryParam("ytToken") @Optional String ytToken,                        
                                        @QueryParam("streamId") String streamId,
                                        @QueryParam("search") String search,
                                        @QueryParam("appPayLoad") String appPayLoad)
            throws Exception {
        String status = "jobId " + jobId + " search " + search;
        List<JobLauncherInfo> results = new ArrayList<JobLauncherInfo>();
        //String env_gobblin_work_dir = System.getenv("GOBBLIN_WORK_DIR");
        String env_gobblin_job_dir = System.getenv("GOBBLIN_JOB_CONFIG_DIR");
        /*
        if(env_gobblin_work_dir == null){
            throw new RuntimeException("Environment property for GOBBLIN_WOParameterRK_DIR is not set.");            
        }
        */
        if (env_gobblin_job_dir == null) {
            throw new RuntimeException("Environment property for GOBBLIN_JOB_CONFIG_DIR is not set.");
        }
        String sysFilePropPath = getConfigDirectory(env_gobblin_job_dir) + "/conf/" + propertiesFile;
        String jobFilePropPath = env_gobblin_job_dir + "/" + jobId + ".pull";

        // Get it from the cache
        Properties sysProp = jobPropertyMap.get(propertiesFile);
        if (sysProp == null) {
            sysProp = ConfigurationConverter.getProperties(new PropertiesConfiguration(sysFilePropPath));
            jobPropertyMap.put(propertiesFile, sysProp);
        }
        // Get it from the cache
        Properties jobProp = jobPropertyMap.get(jobId);
        if (jobProp == null) {
            jobProp = ConfigurationConverter.getProperties(new PropertiesConfiguration(jobFilePropPath));
            jobPropertyMap.put(jobId, jobProp);
        }
        // Not sure if we need this one. This needs to be confirmed in the forums discussion.
        synchronized (this) {
            //ConfigurationConverter.getProperties(new PropertiesConfiguration(sysFilePropPath));
            //ConfigurationConverter.getProperties(new PropertiesConfiguration(jobFilePropPath));
            if (jobProp.get("page.query.parameter") != null) {
                jobProp.put("page.query.parameter", search);
            }

            if (isNotEmpty(streamId)) {
                jobProp.put("stream.id", streamId);
            }
            if (isNotEmpty(fbToken)) {
                jobProp.put("facebook.access.token", fbToken);
            }
            if (isNotEmpty(twToken, twTokenSecret)) {
                jobProp.put("twitter.access.token", twToken);
                jobProp.put("access.token.secret", twTokenSecret);
            }
            if (isNotEmpty(nvToken