Java 类com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient 实例源码

/**
 * Creates a client for accessing Amazon EMR service.
 *
 * @param awsParamsDto the AWS related parameters DTO that includes optional AWS credentials and proxy information
 *
 * @return the Amazon EMR client
 */
@Cacheable(DaoSpringModuleConfig.HERD_CACHE_NAME)
public AmazonElasticMapReduceClient getEmrClient(AwsParamsDto awsParamsDto)
{
    // Get client configuration.
    ClientConfiguration clientConfiguration = awsHelper.getClientConfiguration(awsParamsDto);

    // If specified, use the AWS credentials passed in.
    if (StringUtils.isNotBlank(awsParamsDto.getAwsAccessKeyId()))
    {
        return new AmazonElasticMapReduceClient(
            new BasicSessionCredentials(awsParamsDto.getAwsAccessKeyId(), awsParamsDto.getAwsSecretKey(), awsParamsDto.getSessionToken()),
            clientConfiguration);
    }
    // Otherwise, use the default AWS credentials provider chain.
    else
    {
        return new AmazonElasticMapReduceClient(clientConfiguration);
    }
}

/**
 * Collect data for ElasticMapReduce.
 *
 * @param stats
 *            current statistics object.
 * @param account
 *            currently used credentials object.
 * @param region
 *            currently used aws region.
 */
public static void scanElasticMapReduce(AwsStats stats, AwsAccount account, Regions region) {
    LOG.debug("Scan for MapReduce in region " + region.getName() + " in account " + account.getAccountId());

    try {
        AmazonElasticMapReduce elasticMapReduce = new AmazonElasticMapReduceClient(account.getCredentials());
        elasticMapReduce.setRegion(Region.getRegion(region));

        List<ClusterSummary> list = elasticMapReduce.listClusters().getClusters();

        int totalItems = list.size();
        for (ClusterSummary cs : list) {
            stats.add(new AwsResource(cs.getName(), account.getAccountId(), AwsResourceType.ElasticMapReduce, region));
        }

        LOG.info(totalItems + " ElasticMapReduce clusters in region " + region.getName() + " in account " + account.getAccountId());
    } catch (AmazonServiceException ase) {
        if (ase.getErrorCode().contains("AccessDenied")) {
            LOG.info("Access denied for ElasticMapReduce in region " + region.getName() + " in account " + account.getAccountId());
        } else {
            LOG.error("Exception of ElasticMapReduce: " + ase.getMessage());
        }
    }
}

public void monitorEMRStep() throws Exception {
    List<String> stepIds = new ArrayList<String>();
    Connection conn  = new com.mysql.jdbc.Driver().connect(props.getProperty("url"), props);
    ResultSet openStepsRS = conn.createStatement().executeQuery(props.getProperty("sql.retrieveOpenSteps"));
    AmazonElasticMapReduceClient emr = new AmazonElasticMapReduceClient();
    DescribeStepRequest stepReq=new  DescribeStepRequest();
    PreparedStatement ps = conn.prepareStatement(props.getProperty("sql.updateStepStatus"));
    while(openStepsRS.next()){

        stepReq.setClusterId(openStepsRS.getString("cluster_id"));
        stepReq.setStepId(openStepsRS.getString("step_id"));
        String stepState = emr.describeStep(stepReq).getStep().getStatus().getState();

            if(stepState.equals(StepState.COMPLETED.toString())){
                ps.setString(1,StepState.COMPLETED.toString());
            }else if (stepState.equals(StepState.FAILED.toString())){
                ps.setString(1,StepState.FAILED.toString());                    
            }
            ps.setString(2,openStepsRS.getString("job_config_id"));
            ps.addBatch();
    }

    ps.executeBatch();
    ps.close();
    conn.close();
}

protected List<String> getActiveTaggedClusters() throws Exception{
    AmazonElasticMapReduceClient emrClient = new AmazonElasticMapReduceClient();
    List<String> waitingClusters = new ArrayList<String>();
    ListClustersResult clusterResult = emrClient.listClusters(new ListClustersRequest().withClusterStates(ClusterState.WAITING));

    DescribeClusterRequest specifcTagDescribe = new DescribeClusterRequest();
    specifcTagDescribe.putCustomQueryParameter("Cluster.Tags",null);
     for( ClusterSummary cluster : clusterResult.getClusters()){
            System.out.println("list cluster id "+cluster.getId());
            List<Tag> tagList = emrClient.describeCluster(specifcTagDescribe.withClusterId(cluster.getId())).getCluster().getTags();
            for(Tag tag:tagList){
                if(tag.getKey().equals(props.getProperty("edba.cluster.tag.key"))){
                    waitingClusters.add(cluster.getId());
                }
            }

    }
    return waitingClusters;

}

/**
 * Run the job.
 * @return a the JobFlowId of the job
 */
public String runJob() {

  // Get the credentials
  final AWSCredentials credentials =
      new BasicAWSCredentials(this.AWSAccessKey, this.AWSSecretKey);

  // Create the Amazon Elastic MapReduce object
  this.elasticMapReduceClient = new AmazonElasticMapReduceClient(credentials);

  // Set the end point
  this.elasticMapReduceClient.setEndpoint(this.endpoint);

  this.runFlowResult =
      this.elasticMapReduceClient.runJobFlow(this.runFlowRequest);

  return this.runFlowResult.getJobFlowId();
}

/**
 * Terminate EMR cluster, overrides terminate protection if requested.
 */
@Override
public void terminateEmrCluster(AmazonElasticMapReduceClient emrClient, String clusterId, boolean overrideTerminationProtection)
{
    // Override terminate protection if requested.
    if (overrideTerminationProtection)
    {
        // Set termination protection
        emrClient.setTerminationProtection(new SetTerminationProtectionRequest().withJobFlowIds(clusterId).withTerminationProtected(false));
    }

    // Terminate the job flow
    emrClient.terminateJobFlows(new TerminateJobFlowsRequest().withJobFlowIds(clusterId));
}

@Test
public void testGetEmrClientCacheHitMiss()
{
    // Create an AWS parameters DTO that contains both AWS credentials and proxy information.
    AwsParamsDto awsParamsDto =
        new AwsParamsDto(AWS_ASSUMED_ROLE_ACCESS_KEY, AWS_ASSUMED_ROLE_SECRET_KEY, AWS_ASSUMED_ROLE_SESSION_TOKEN, HTTP_PROXY_HOST, HTTP_PROXY_PORT);

    // Get an Amazon EMR client.
    AmazonElasticMapReduceClient amazonElasticMapReduceClient = awsClientFactory.getEmrClient(awsParamsDto);

    // Confirm a cache hit.
    assertEquals(amazonElasticMapReduceClient, awsClientFactory.getEmrClient(
        new AwsParamsDto(AWS_ASSUMED_ROLE_ACCESS_KEY, AWS_ASSUMED_ROLE_SECRET_KEY, AWS_ASSUMED_ROLE_SESSION_TOKEN, HTTP_PROXY_HOST, HTTP_PROXY_PORT)));

    // Confirm a cache miss due to AWS credentials.
    assertNotEquals(amazonElasticMapReduceClient, awsClientFactory.getEmrClient(
        new AwsParamsDto(AWS_ASSUMED_ROLE_ACCESS_KEY_2, AWS_ASSUMED_ROLE_SECRET_KEY_2, AWS_ASSUMED_ROLE_SESSION_TOKEN_2, HTTP_PROXY_HOST,
            HTTP_PROXY_PORT)));

    // Confirm a cache miss due to http proxy information.
    assertNotEquals(amazonElasticMapReduceClient, awsClientFactory.getEmrClient(
        new AwsParamsDto(AWS_ASSUMED_ROLE_ACCESS_KEY, AWS_ASSUMED_ROLE_SECRET_KEY, AWS_ASSUMED_ROLE_SESSION_TOKEN, HTTP_PROXY_HOST_2, HTTP_PROXY_PORT_2)));

    // Clear the cache.
    cacheManager.getCache(DaoSpringModuleConfig.HERD_CACHE_NAME).clear();

    // Confirm a cache miss due to cleared cache.
    assertNotEquals(amazonElasticMapReduceClient, awsClientFactory.getEmrClient(awsParamsDto));
}

@Test
public void getEmrClientAssertClientConfigurationSet() throws Exception
{
    String httpProxyHost = "httpProxyHost";
    Integer httpProxyPort = 1234;

    AwsParamsDto awsParamsDto = new AwsParamsDto();
    awsParamsDto.setHttpProxyHost(httpProxyHost);
    awsParamsDto.setHttpProxyPort(httpProxyPort);
    AmazonElasticMapReduceClient amazonElasticMapReduceClient = emrDao.getEmrClient(awsParamsDto);
    ClientConfiguration clientConfiguration = (ClientConfiguration) ReflectionTestUtils.getField(amazonElasticMapReduceClient, "clientConfiguration");
    assertNotNull(clientConfiguration);
    assertEquals(httpProxyHost, clientConfiguration.getProxyHost());
    assertEquals(httpProxyPort.intValue(), clientConfiguration.getProxyPort());
}

@Test
public void getEmrClientAssertClientConfigurationNotSetWhenProxyHostIsBlank() throws Exception
{
    String httpProxyHost = "";
    Integer httpProxyPort = 1234;

    AwsParamsDto awsParamsDto = new AwsParamsDto();
    awsParamsDto.setHttpProxyHost(httpProxyHost);
    awsParamsDto.setHttpProxyPort(httpProxyPort);
    AmazonElasticMapReduceClient amazonElasticMapReduceClient = emrDao.getEmrClient(awsParamsDto);
    ClientConfiguration clientConfiguration = (ClientConfiguration) ReflectionTestUtils.getField(amazonElasticMapReduceClient, "clientConfiguration");
    assertNotNull(clientConfiguration);
    assertNull(clientConfiguration.getProxyHost());
}

@Test
public void getEmrClientAssertClientConfigurationNotSetWhenProxyPortIsNull() throws Exception
{
    String httpProxyHost = "httpProxyHost";
    Integer httpProxyPort = null;

    AwsParamsDto awsParamsDto = new AwsParamsDto();
    awsParamsDto.setHttpProxyHost(httpProxyHost);
    awsParamsDto.setHttpProxyPort(httpProxyPort);
    AmazonElasticMapReduceClient amazonElasticMapReduceClient = emrDao.getEmrClient(awsParamsDto);
    ClientConfiguration clientConfiguration = (ClientConfiguration) ReflectionTestUtils.getField(amazonElasticMapReduceClient, "clientConfiguration");
    assertNotNull(clientConfiguration);
    assertNull(clientConfiguration.getProxyHost());
}

@Override
public ListClustersResult listEmrClusters(AmazonElasticMapReduceClient emrClient, ListClustersRequest listClustersRequest)
{
    List<ClusterSummary> clusterSummaryList = new ArrayList<>();
    for (MockEmrJobFlow cluster : emrClusters.values())
    {
        if (!listClustersRequest.getClusterStates().isEmpty() && listClustersRequest.getClusterStates().contains(cluster.getStatus()))
        {
            ClusterSummary clusterSummary = new ClusterSummary();
            clusterSummary.withId(cluster.getJobFlowId()).withName(cluster.getJobFlowName()).withStatus(new ClusterStatus().withState(cluster.getStatus())
                .withStateChangeReason(new ClusterStateChangeReason().withCode(cluster.getStatusChangeReason().getCode())
                    .withMessage(cluster.getStatusChangeReason().getMessage())).withTimeline(new ClusterTimeline().withCreationDateTime(
                    cluster.getStatusTimeline().getCreationTime() != null ? cluster.getStatusTimeline().getCreationTime().toGregorianCalendar().getTime() :
                        null).withEndDateTime(
                    cluster.getStatusTimeline().getEndTime() != null ? cluster.getStatusTimeline().getEndTime().toGregorianCalendar().getTime() : null)
                    .withReadyDateTime(
                        cluster.getStatusTimeline().getReadyTime() != null ? cluster.getStatusTimeline().getReadyTime().toGregorianCalendar().getTime() :
                            null)));
            clusterSummaryList.add(clusterSummary);
        }
    }
    if (StringUtils.isBlank(listClustersRequest.getMarker()))
    {
        return new ListClustersResult().withClusters(clusterSummaryList).withMarker(MOCK_EMR_MAKER);
    }
    else
    {
        return new ListClustersResult().withClusters(clusterSummaryList);
    }
}

@Override
public ListInstancesResult listClusterInstancesRequest(AmazonElasticMapReduceClient emrClient, ListInstancesRequest listInstancesRequest)
{
    MockEmrJobFlow cluster =
        getClusterByName(buildEmrClusterName(AbstractDaoTest.NAMESPACE, AbstractDaoTest.EMR_CLUSTER_DEFINITION_NAME, MOCK_CLUSTER_NOT_PROVISIONED_NAME));

    if (cluster != null && listInstancesRequest.getClusterId().equals(cluster.getJobFlowId()))
    {
        return new ListInstancesResult();
    }
    Instance instance = new Instance().withEc2InstanceId("EC2_EMR_MASTER_INSTANCE").withPrivateIpAddress("INSTANCE_IP_ADDRESS");
    return new ListInstancesResult().withInstances(instance);
}

@Override
public void terminateEmrCluster(AmazonElasticMapReduceClient emrClient, String clusterId, boolean overrideTerminationProtection)
{
    MockEmrJobFlow cluster = getClusterById(clusterId);
    if (cluster.getJobFlowName().endsWith(MockAwsOperationsHelper.AMAZON_SERVICE_EXCEPTION))
    {
        throw new AmazonServiceException(MockAwsOperationsHelper.AMAZON_SERVICE_EXCEPTION);
    }
    cluster.setStatus(ClusterState.TERMINATED.toString());
}

@Override
public ListInstanceFleetsResult listInstanceFleets(AmazonElasticMapReduceClient emrClient, ListInstanceFleetsRequest listInstanceFleetsRequest)
{
    ListInstanceFleetsResult listInstanceFleetsResult = new ListInstanceFleetsResult();
    List<InstanceFleet> instanceFleets = new ArrayList<>();
    InstanceFleet instanceFleet = new InstanceFleet();
    instanceFleet.setId("mock_instance_id_1");
    instanceFleet.setName("mock_instance_name");
    instanceFleets.add(instanceFleet);
    listInstanceFleetsResult.setInstanceFleets(instanceFleets);
    return listInstanceFleetsResult;
}

@Test
public void testGetListInstanceFleetsResult()
{
    // Create an AWS parameters DTO.
    AwsParamsDto awsParamsDto =
        new AwsParamsDto(AWS_ASSUMED_ROLE_ACCESS_KEY, AWS_ASSUMED_ROLE_SECRET_KEY, AWS_ASSUMED_ROLE_SESSION_TOKEN, HTTP_PROXY_HOST, HTTP_PROXY_PORT);

    // Create a mock AmazonElasticMapReduceClient.
    AmazonElasticMapReduceClient amazonElasticMapReduceClient = mock(AmazonElasticMapReduceClient.class);

    // Create a list instance fleets request.
    ListInstanceFleetsRequest listInstanceFleetsRequest = new ListInstanceFleetsRequest().withClusterId(EMR_CLUSTER_ID);

    // Create a list instance fleets result.
    ListInstanceFleetsResult listInstanceFleetsResult = new ListInstanceFleetsResult().withMarker(MARKER);

    // Mock the external calls.
    when(awsClientFactory.getEmrClient(awsParamsDto)).thenReturn(amazonElasticMapReduceClient);
    when(emrOperations.listInstanceFleets(amazonElasticMapReduceClient, listInstanceFleetsRequest)).thenReturn(listInstanceFleetsResult);

    // Call the method under test.
    ListInstanceFleetsResult result = emrDaoImpl.getListInstanceFleetsResult(EMR_CLUSTER_ID, awsParamsDto);

    // Verify the external calls.
    verify(awsClientFactory).getEmrClient(awsParamsDto);
    verify(emrOperations).listInstanceFleets(amazonElasticMapReduceClient, listInstanceFleetsRequest);
    verifyNoMoreInteractionsHelper();

    // Validate the results.
    assertEquals(listInstanceFleetsResult, result);
}

/**
 * Helper class to create and Amazon EMR cluster with HBase installed on that cluster
 * 
 * @param clusterIdentifier - cluster id if one exists
 * @param clusterName - name you want associated with this cluster
 * @param amiVersion - version of AMI that you wish to use for your HBase cluster
 * @param keypair - you need a keypair to SSH into the cluster
 * @param masterInstanceType - Amazon EC2 instance type for your master node
 * @param coreInstanceType - Amazon Ec2 instance tyoe for your core nodes
 * @param logUri - Specify a bucket for your EMR logs
 * @param numberOfNodes - total number of nodes in your cluster including the master node
 */

private void createEMRCluster(String clusterIdentifier,
        String clusterName,
        String amiVersion,
        String keypair,
        String masterInstanceType,
        String coreInstanceType,
        String logUri,
        int numberOfNodes) {
    // Make sure the EMR cluster is available
    AmazonElasticMapReduceClient emrClient = new AmazonElasticMapReduceClient(config.AWS_CREDENTIALS_PROVIDER);
    emrClient.setEndpoint(config.EMR_ENDPOINT);
    String clusterid = clusterIdentifier;
    if (!EMRUtils.clusterExists(emrClient, clusterIdentifier)) {
        clusterid = EMRUtils.createCluster(emrClient,
                    clusterIdentifier,
                    amiVersion,
                    keypair,
                    masterInstanceType,
                    coreInstanceType,
                    logUri,
                    numberOfNodes);
    }
    // Update the emr cluster id and public DNS properties
    config.EMR_CLUSTER_IDENTIFIER = clusterid;
    config.EMR_CLUSTER_PUBLIC_DNS = EMRUtils.getPublicDns(emrClient, clusterid);            
    //make sure table exists
    if (!HBaseUtils.tablesExists(config.HBASE_TABLE_NAME, config.EMR_CLUSTER_PUBLIC_DNS, config.HBASE_REST_PORT)){
        HBaseUtils.createTable(config.HBASE_TABLE_NAME, config.EMR_CLUSTER_PUBLIC_DNS, config.HBASE_REST_PORT); 
    }

}

public HBaseEmitter(EMRHBaseKinesisConnectorConfiguration configuration) {
    // DynamoDB Config
    this.emrEndpoint = configuration.EMR_ENDPOINT;
    this.hbaseTableName = configuration.HBASE_TABLE_NAME;
    this.hbaseRestPort = configuration.HBASE_REST_PORT;
    this.emrPublicDns = configuration.EMR_CLUSTER_PUBLIC_DNS;
    // Client
    this.emrClient = new AmazonElasticMapReduceClient(configuration.AWS_CREDENTIALS_PROVIDER);
    this.emrClient.setEndpoint(this.emrEndpoint);
    LOG.info("EMRHBaseEmitter.....");
}

protected String fireEMRJob(String paramsStr,String clusterId){
    StepFactory stepFactory = new StepFactory();
    AmazonElasticMapReduceClient emr = new AmazonElasticMapReduceClient();
    emr.setRegion(Region.getRegion(Regions.fromName(System.getenv().get("AWS_REGION"))));
    Application sparkConfig = new Application()
            .withName("Spark");

    String[] params = paramsStr.split(",");
    StepConfig enabledebugging = new StepConfig()
            .withName("Enable debugging")
            .withActionOnFailure("TERMINATE_JOB_FLOW")
            .withHadoopJarStep(stepFactory.newEnableDebuggingStep());

    HadoopJarStepConfig sparkStepConf = new HadoopJarStepConfig()
            .withJar("command-runner.jar")
            .withArgs(params);  

    final StepConfig sparkStep = new StepConfig()
            .withName("Spark Step")
            .withActionOnFailure("CONTINUE")
            .withHadoopJarStep(sparkStepConf);


    AddJobFlowStepsRequest request = new AddJobFlowStepsRequest(clusterId)
            .withSteps(new ArrayList<StepConfig>(){{add(sparkStep);}});


    AddJobFlowStepsResult result = emr.addJobFlowSteps(request);
    return result.getStepIds().get(0);
}

@Override
public AmazonElasticMapReduceClient getEmrClient(AwsParamsDto awsParamsDto)
{
    return awsClientFactory.getEmrClient(awsParamsDto);
}

/**
 * List EMR cluster instances
 */
@Override
public ListInstancesResult listClusterInstancesRequest(AmazonElasticMapReduceClient emrClient, ListInstancesRequest listInstancesRequest)
{
    return emrClient.listInstances(listInstancesRequest);
}

/**
 * Run Job Flow to AmazonElasticMapReduceClient
 */
@Override
public String runEmrJobFlow(AmazonElasticMapReduceClient emrClient, RunJobFlowRequest jobFlowRequest)
{
    return emrClient.runJobFlow(jobFlowRequest).getJobFlowId();
}

/**
 * List the EMR Clusters in the account
 */
@Override
public ListClustersResult listEmrClusters(AmazonElasticMapReduceClient emrClient, ListClustersRequest listClustersRequest)
{
    return emrClient.listClusters(listClustersRequest);
}

@Override
public ListInstanceFleetsResult listInstanceFleets(AmazonElasticMapReduceClient emrClient, ListInstanceFleetsRequest listInstanceFleetsRequest)
{
    return emrClient.listInstanceFleets(listInstanceFleetsRequest);
}

@Override
public String runEmrJobFlow(AmazonElasticMapReduceClient emrClient, RunJobFlowRequest jobFlowRequest)
{
    String clusterStatus = ClusterState.BOOTSTRAPPING.toString();

    StatusChangeReason reason = new StatusChangeReason(ClusterStateChangeReasonCode.USER_REQUEST.toString(), "Started " + clusterStatus);
    StatusTimeline timeline = new StatusTimeline();
    timeline.setCreationTime(HerdDateUtils.getXMLGregorianCalendarValue(new Date()));

    if (StringUtils.isNotBlank(jobFlowRequest.getAmiVersion()))
    {
        if (jobFlowRequest.getAmiVersion().equals(MockAwsOperationsHelper.AMAZON_THROTTLING_EXCEPTION))
        {
            AmazonServiceException throttlingException = new AmazonServiceException("test throttling exception");
            throttlingException.setErrorCode("ThrottlingException");

            throw throttlingException;
        }
        else if (jobFlowRequest.getAmiVersion().equals(MockAwsOperationsHelper.AMAZON_BAD_REQUEST))
        {
            AmazonServiceException badRequestException = new AmazonServiceException(MockAwsOperationsHelper.AMAZON_BAD_REQUEST);
            badRequestException.setStatusCode(HttpStatus.SC_BAD_REQUEST);
            throw badRequestException;
        }
        else if (jobFlowRequest.getAmiVersion().equals(MockAwsOperationsHelper.AMAZON_NOT_FOUND))
        {
            AmazonServiceException notFoundException = new AmazonServiceException(MockAwsOperationsHelper.AMAZON_NOT_FOUND);
            notFoundException.setStatusCode(HttpStatus.SC_NOT_FOUND);
            throw notFoundException;
        }
        else if (jobFlowRequest.getAmiVersion().equals(MockAwsOperationsHelper.AMAZON_SERVICE_EXCEPTION))
        {
            throw new AmazonServiceException(MockAwsOperationsHelper.AMAZON_SERVICE_EXCEPTION);
        }
        else if (jobFlowRequest.getAmiVersion().equals(MockAwsOperationsHelper.AMAZON_CLUSTER_STATUS_WAITING))
        {
            clusterStatus = ClusterState.WAITING.toString();
        }
        else if (jobFlowRequest.getAmiVersion().equals(MockAwsOperationsHelper.AMAZON_CLUSTER_STATUS_RUNNING))
        {
            clusterStatus = ClusterState.RUNNING.toString();
        }
    }


    return createNewCluster(jobFlowRequest, clusterStatus, reason, timeline).getJobFlowId();
}

@Test
public void testGetActiveEmrClusterByName()
{
    // Create an AWS parameters DTO.
    AwsParamsDto awsParamsDto =
        new AwsParamsDto(AWS_ASSUMED_ROLE_ACCESS_KEY, AWS_ASSUMED_ROLE_SECRET_KEY, AWS_ASSUMED_ROLE_SESSION_TOKEN, HTTP_PROXY_HOST, HTTP_PROXY_PORT);

    // Create a mock AmazonElasticMapReduceClient.
    AmazonElasticMapReduceClient amazonElasticMapReduceClient = mock(AmazonElasticMapReduceClient.class);

    // Create a list cluster request.
    ListClustersRequest listClustersRequest = new ListClustersRequest().withClusterStates(EMR_VALID_STATE);

    // Create a list cluster result with a non-matching cluster and a marker.
    ListClustersResult listClusterResultWithMarker = new ListClustersResult().withClusters(new ClusterSummary().withName(INVALID_VALUE)).withMarker(MARKER);

    // Create a list cluster request with marker.
    ListClustersRequest listClustersRequestWithMarker = new ListClustersRequest().withClusterStates(EMR_VALID_STATE).withMarker(MARKER);

    // Create a cluster summary.
    ClusterSummary clusterSummary = new ClusterSummary().withName(EMR_CLUSTER_NAME);

    // Create a list cluster result with the matching cluster.
    ListClustersResult listClusterResult = new ListClustersResult().withClusters(clusterSummary);

    // Mock the external calls.
    when(configurationHelper.getProperty(ConfigurationValue.EMR_VALID_STATES)).thenReturn(EMR_VALID_STATE);
    when(configurationHelper.getProperty(ConfigurationValue.FIELD_DATA_DELIMITER))
        .thenReturn((String) ConfigurationValue.FIELD_DATA_DELIMITER.getDefaultValue());
    when(awsClientFactory.getEmrClient(awsParamsDto)).thenReturn(amazonElasticMapReduceClient);
    when(emrOperations.listEmrClusters(amazonElasticMapReduceClient, listClustersRequest)).thenReturn(listClusterResultWithMarker);
    when(emrOperations.listEmrClusters(amazonElasticMapReduceClient, listClustersRequestWithMarker)).thenReturn(listClusterResult);

    // Call the method under test.
    ClusterSummary result = emrDaoImpl.getActiveEmrClusterByName(EMR_CLUSTER_NAME, awsParamsDto);

    // Verify the external calls.
    verify(configurationHelper).getProperty(ConfigurationValue.EMR_VALID_STATES);
    verify(configurationHelper).getProperty(ConfigurationValue.FIELD_DATA_DELIMITER);
    verify(awsClientFactory, times(2)).getEmrClient(awsParamsDto);
    verify(emrOperations, times(2)).listEmrClusters(eq(amazonElasticMapReduceClient), any(ListClustersRequest.class));
    verifyNoMoreInteractionsHelper();

    // Validate the results.
    assertEquals(clusterSummary, result);
}

@Override
public AmazonElasticMapReduceClient getEmrClient(AwsParamsDto awsParamsDto)
{
     return super.getEmrClient(awsParamsDto);
}

/**
 * Create the EMR client with the given proxy and access key details.
 *
 * @param awsParamsDto AWS related parameters for access/secret keys and proxy details.
 *
 * @return the AmazonElasticMapReduceClient object.
 */
public AmazonElasticMapReduceClient getEmrClient(AwsParamsDto awsParamsDto);

public String runEmrJobFlow(AmazonElasticMapReduceClient emrClient, RunJobFlowRequest jobFlowRequest);

public ListClustersResult listEmrClusters(AmazonElasticMapReduceClient emrClient, ListClustersRequest listClustersRequest);

public ListInstancesResult listClusterInstancesRequest(AmazonElasticMapReduceClient emrClient, ListInstancesRequest listInstancesRequest);

public void terminateEmrCluster(AmazonElasticMapReduceClient emrClient, String clusterId, boolean overrideTerminationProtection);

public ListInstanceFleetsResult listInstanceFleets(AmazonElasticMapReduceClient emrClient, ListInstanceFleetsRequest listInstanceFleetsRequest);