class MapReduceBean {

    @POST
    @Produces({"application/xml", "text/xml", "application/json", "text/yaml", "text/x-yaml", "application/x-yaml", "application/x-protobuf",
            "application/x-protostuff"})
    @javax.ws.rs.Path("/submit")
    @GZIP
    public GenericResponse<String> submit(@FormParam("jobName") String jobName, @FormParam("parameters") String parameters) {
        GenericResponse<String> response = new GenericResponse<>();
        
        // Find out who/what called this method
        Principal p = ctx.getCallerPrincipal();
        String sid;
        Set<Collection<String>> cbAuths = new HashSet<>();
        DatawavePrincipal datawavePrincipal = null;
        
        if (p instanceof DatawavePrincipal) {
            datawavePrincipal = (DatawavePrincipal) p;
            sid = datawavePrincipal.getShortName();
            cbAuths.addAll(datawavePrincipal.getAuthorizations());
        } else {
            QueryException qe = new QueryException(DatawaveErrorCode.UNEXPECTED_PRINCIPAL_ERROR, MessageFormat.format("Class: {0}", p.getClass().getName()));
            response.addException(qe);
            throw new DatawaveWebApplicationException(qe, response);
        }
        
        // Get the MapReduceJobConfiguration from the configuration
        MapReduceJobConfiguration job;
        try {
            job = this.mapReduceConfiguration.getConfiguration(jobName);
        } catch (IllegalArgumentException e) {
            BadRequestQueryException qe = new BadRequestQueryException(DatawaveErrorCode.JOB_CONFIGURATION_ERROR, e);
            response.addException(qe);
            throw new BadRequestException(qe, response);
        }
        
        // Ensure that the user has the required roles and has passed the required auths
        if (null != job.getRequiredRoles() || null != job.getRequiredAuths()) {
            try {
                canRunJob(datawavePrincipal, new MultivaluedMapImpl<>(), job.getRequiredRoles(), job.getRequiredAuths());
            } catch (UnauthorizedQueryException qe) {
                // user does not have all of the required roles or did not pass the required auths
                response.addException(qe);
                throw new UnauthorizedException(qe, response);
            }
        }
        
        // Parse the parameters
        Map<String,String> runtimeParameters = new HashMap<>();
        if (null != parameters) {
            String[] param = parameters.split(PARAMETER_SEPARATOR);
            for (String yyy : param) {
                String[] parts = yyy.split(PARAMETER_NAME_VALUE_SEPARATOR);
                if (parts.length == 2) {
                    runtimeParameters.put(parts[0], parts[1]);
                }
            }
        }
        
        // Check to see if the job configuration class implements specific interfaces.
        if (job instanceof NeedCallerDetails) {
            ((NeedCallerDetails) job).setUserSid(sid);
            ((NeedCallerDetails) job).setPrincipal(p);
        }
        if (job instanceof NeedAccumuloConnectionFactory) {
            ((NeedAccumuloConnectionFactory) job).setAccumuloConnectionFactory(this.connectionFactory);
        }
        if (job instanceof NeedAccumuloDetails) {
            ((NeedAccumuloDetails) job).setUsername(this.connectionPoolsConfiguration.getPools().get(this.connectionPoolsConfiguration.getDefaultPool())
                            .getUsername());
            ((NeedAccumuloDetails) job).setPassword(this.connectionPoolsConfiguration.getPools().get(this.connectionPoolsConfiguration.getDefaultPool())
                            .getPassword());
            ((NeedAccumuloDetails) job).setInstanceName(this.connectionPoolsConfiguration.getPools().get(this.connectionPoolsConfiguration.getDefaultPool())
                            .getInstance());
            ((NeedAccumuloDetails) job).setZookeepers(this.connectionPoolsConfiguration.getPools().get(this.connectionPoolsConfiguration.getDefaultPool())
                            .getZookeepers());
        }
        if (job instanceof NeedQueryLogicFactory) {
            ((NeedQueryLogicFactory) job).setQueryLogicFactory(this.queryLogicFactory);
        }
        if (job instanceof NeedQueryPersister) {
            ((NeedQueryPersister) job).setPersister(this.queryPersister);
        }
        
        if (job instanceof NeedQueryCache) {
            ((NeedQueryCache) job).setQueryCache(cache);
        }
        
        if (job instanceof NeedSecurityDomain) {
            ((NeedSecurityDomain) job).setSecurityDomain(this.jsseSecurityDomain);
        }
        
        // If this job is being restarted, then the jobId will be the same. The restart method
        // puts the id into the runtime parameters
        String id = runtimeParameters.get(JOB_ID);
        if (null == id)
            id = UUID.randomUUID().toString();
        org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
        
        StringBuilder name = new StringBuilder().append(jobName).append("_sid_").append(sid).append("_id_").append(id);
        Job j;
        try {
            j = createJob(conf, name);
            job.initializeConfiguration(id, j, runtimeParameters, serverPrincipal);
        } catch (WebApplicationException waEx) {
            throw waEx;
        } catch (Exception e) {
            QueryException qe = new QueryException(DatawaveErrorCode.LOGIC_CONFIGURATION_ERROR, e);
            log.error(qe.getMessage(), e);
            response.addException(qe.getBottomQueryException());
            throw new DatawaveWebApplicationException(qe, response);
        }
        
        // Enforce that certain InputFormat classes are being used here.
        if (this.mapReduceConfiguration.isRestrictInputFormats()) {
            // Make sure that the job input format is in the list
            Class<? extends InputFormat<?,?>> ifClass;
            try {
                ifClass = j.getInputFormatClass();
            } catch (ClassNotFoundException e1) {
                QueryException qe = new QueryException(DatawaveErrorCode.INPUT_FORMAT_CLASS_ERROR, e1);
                log.error(qe);
                response.addException(qe);
                throw new DatawaveWebApplicationException(qe, response);
            }
            if (!this.mapReduceConfiguration.getValidInputFormats().contains(ifClass)) {
                IllegalArgumentException e = new IllegalArgumentException("Invalid input format class specified. Must use one of "
                                + this.mapReduceConfiguration.getValidInputFormats());
                QueryException qe = new QueryException(DatawaveErrorCode.INVALID_FORMAT, e);
                log.error(qe);
                response.addException(qe.getBottomQueryException());
                throw new DatawaveWebApplicationException(qe, response);
            }
        }
        
        try {
            j.submit();
        } catch (Exception e) {
            QueryException qe = new QueryException(DatawaveErrorCode.MAPREDUCE_JOB_START_ERROR, e);
            log.error(qe.getMessage(), qe);
            response.addException(qe.getBottomQueryException());
            throw new DatawaveWebApplicationException(qe, response);
        }
        
        JobID mapReduceJobId = j.getJobID();
        log.info("JOB ID: " + mapReduceJobId);
        
        // Create an entry in the state table
        boolean restarted = (runtimeParameters.get(JOB_ID) != null);
        try {
            if (!restarted)
                mapReduceState.create(id, job.getHdfsUri(), job.getJobTracker(), job.getJobDir(), mapReduceJobId.toString(), job.getResultsDir(), parameters,
                                jobName);
            else
                mapReduceState.addJob(id, mapReduceJobId.toString());
        } catch (Exception e) {
            QueryException qe = new QueryException(DatawaveErrorCode.MAPREDUCE_STATE_PERSISTENCE_ERROR, e);
            log.error(qe);
            response.addException(qe.getBottomQueryException());
            try {
                j.killJob();
            } catch (IOException ioe) {
                QueryException qe2 = new QueryException(DatawaveErrorCode.MAPREDUCE_JOB_KILL_ERROR, ioe);
                response.addException(qe2);
            }
            throw new DatawaveWebApplicationException(qe, response);
            
        }
        response.setResult(id);
        return response;
        
    }

}