[ML] Handle failures in the idiomatic way (elastic/x-pack-elasticsearch#1785)

This commit changes a couple of places where our ExceptionsHelper
class was throwing exceptions to instead return the exceptions.
Then they can be passed to onFailure() methods or thrown depending
on what's appropriate for the caller.  This is the standard Elastic
way of handling failures.

Original commit: elastic/x-pack-elasticsearch@fce07eb075

plugin/src/main/java/org/elasticsearch/xpack/ml/job/JobManager.java

@@ -167,6 +167,7 @@ public class JobManager extends AbstractComponent {
         MlMetadata currentMlMetadata = state.metaData().custom(MlMetadata.TYPE);
         if (currentMlMetadata != null && currentMlMetadata.getJobs().containsKey(job.getId())) {
             actionListener.onFailure(ExceptionsHelper.jobAlreadyExists(job.getId()));
+            return;
         }
 
         ActionListener<Boolean> putJobListener = new ActionListener<Boolean>() {

plugin/src/main/java/org/elasticsearch/xpack/ml/utils/ExceptionsHelper.java

@@ -23,11 +23,11 @@ public class ExceptionsHelper {
     }
 
     public static ResourceAlreadyExistsException jobAlreadyExists(String jobId) {
-        throw new ResourceAlreadyExistsException(Messages.getMessage(Messages.JOB_CONFIG_ID_ALREADY_TAKEN, jobId));
+        return new ResourceAlreadyExistsException(Messages.getMessage(Messages.JOB_CONFIG_ID_ALREADY_TAKEN, jobId));
     }
 
     public static ResourceNotFoundException missingDatafeedException(String datafeedId) {
-        throw new ResourceNotFoundException(Messages.getMessage(Messages.DATAFEED_NOT_FOUND, datafeedId));
+        return new ResourceNotFoundException(Messages.getMessage(Messages.DATAFEED_NOT_FOUND, datafeedId));
     }
 
     public static ElasticsearchException serverError(String msg) {

plugin/src/test/java/org/elasticsearch/xpack/ml/job/JobManagerTests.java

@@ -167,18 +167,17 @@ public class JobManagerTests extends ESTestCase {
         ClusterState clusterState = ClusterState.builder(new ClusterName("name"))
                 .metaData(MetaData.builder().putCustom(MlMetadata.TYPE, mlMetadata.build())).build();
 
-        expectThrows(ResourceAlreadyExistsException.class, () ->
         jobManager.putJob(putJobRequest, clusterState, new ActionListener<PutJobAction.Response>() {
             @Override
             public void onResponse(PutJobAction.Response response) {
-
+                fail("should have got an error");
             }
 
             @Override
             public void onFailure(Exception e) {
-                            fail(e.toString());
+                assertTrue(e instanceof ResourceAlreadyExistsException);
             }
-                    }));
+        });
     }
 
     private Job.Builder createJob() {