### Question:
TaskAttemptResourceProvider extends
    AbstractJDBCResourceProvider<TaskAttemptResourceProvider.TaskAttemptFields> { @Override public Set<Resource> getResources(Request request, Predicate predicate) throws SystemException, UnsupportedPropertyException, NoSuchResourceException, NoSuchParentResourceException { Set<Resource> resourceSet = new HashSet<>(); Set<String> requestedIds = getRequestPropertyIds(request, predicate); Set<Map<String,Object>> predicatePropertieSet = getPropertyMaps(predicate); for (Map<String,Object> predicateProperties : predicatePropertieSet) { String clusterName = (String) predicateProperties .get(TASK_ATTEMPT_CLUSTER_NAME_PROPERTY_ID); String workflowId = (String) predicateProperties .get(TASK_ATTEMPT_WORKFLOW_ID_PROPERTY_ID); String jobId = (String) predicateProperties .get(TASK_ATTEMPT_JOB_ID_PROPERTY_ID); String taskAttemptId = (String) predicateProperties .get(TASK_ATTEMPT_ID_PROPERTY_ID); resourceSet.addAll(taskAttemptFetcher.fetchTaskAttemptDetails( requestedIds, clusterName, workflowId, jobId, taskAttemptId)); } return resourceSet; } protected  TaskAttemptResourceProvider(); protected  TaskAttemptResourceProvider(Set<String> propertyIds,
      Map<Type,String> keyPropertyIds, TaskAttemptFetcher taskAttemptFetcher); @Override RequestStatus createResources(Request request); @Override Set<Resource> getResources(Request request, Predicate predicate); @Override RequestStatus updateResources(Request request, Predicate predicate); @Override RequestStatus deleteResources(Request request, Predicate predicate); @Override Map<Type,String> getKeyPropertyIds();  }### Answer:
@Test public void testGetResources() throws SystemException, UnsupportedPropertyException, NoSuchResourceException, NoSuchParentResourceException { Set<Resource> expected = new HashSet<>(); expected.add(createTaskAttemptResponse("Cluster100", "workflow1", "job1", "taskAttempt1")); expected.add(createTaskAttemptResponse("Cluster100", "workflow2", "job2", "taskAttempt2")); expected.add(createTaskAttemptResponse("Cluster100", "workflow2", "job2", "taskAttempt3")); Resource.Type type = Resource.Type.TaskAttempt; Set<String> propertyIds = TaskAttemptResourceProvider.propertyIds; TaskAttemptFetcher taskAttemptFetcher = createMock(TaskAttemptFetcher.class); expect( taskAttemptFetcher.fetchTaskAttemptDetails(propertyIds, null, null, "job2", null)).andReturn(expected).once(); replay(taskAttemptFetcher); Map<Resource.Type,String> keyPropertyIds = PropertyHelper .getKeyPropertyIds(type); ResourceProvider provider = new TaskAttemptResourceProvider(propertyIds, keyPropertyIds, taskAttemptFetcher); Request request = PropertyHelper.getReadRequest(propertyIds); Predicate predicate = new PredicateBuilder() .property(TaskAttemptResourceProvider.TASK_ATTEMPT_JOB_ID_PROPERTY_ID) .equals("job2").toPredicate(); Set<Resource> resources = provider.getResources(request, predicate); Assert.assertEquals(3, resources.size()); Set<String> names = new HashSet<>(); for (Resource resource : resources) { String clusterName = (String) resource .getPropertyValue(TaskAttemptResourceProvider.TASK_ATTEMPT_CLUSTER_NAME_PROPERTY_ID); Assert.assertEquals("Cluster100", clusterName); names .add((String) resource .getPropertyValue(TaskAttemptResourceProvider.TASK_ATTEMPT_ID_PROPERTY_ID)); } for (Resource resource : expected) { Assert .assertTrue(names.contains(resource .getPropertyValue(TaskAttemptResourceProvider.TASK_ATTEMPT_ID_PROPERTY_ID))); } verify(taskAttemptFetcher); }