public static List<ComponentContainers> filterInstances(
      ServiceContext context,
      ClientAMProtocol.GetCompInstancesRequestProto filterReq) {
    Map<String, ComponentContainers> containersByComp = new HashMap<>();

    Map<ContainerId, ComponentInstance> instances =
        context.scheduler.getLiveInstances();

    instances.forEach(((containerId, instance) -> {
      boolean include = true;
      if (filterReq.getComponentNamesList() != null &&
          !filterReq.getComponentNamesList().isEmpty()) {
        // filter by component name
        if (!filterReq.getComponentNamesList().contains(
            instance.getComponent().getName())) {
          include = false;
        }
      }

      if (filterReq.getVersion() != null && !filterReq.getVersion().isEmpty()) {
        // filter by version
        String instanceServiceVersion = instance.getServiceVersion();
        if (instanceServiceVersion == null || !instanceServiceVersion.equals(
            filterReq.getVersion())) {
          include = false;
        }
      }

      if (filterReq.getContainerStatesList() != null &&
          !filterReq.getContainerStatesList().isEmpty()) {
        // filter by state
        if (!filterReq.getContainerStatesList().contains(
            instance.getContainerState().toString())) {
          include = false;
        }
      }

      if (include) {
        ComponentContainers compContainers =
            containersByComp.computeIfAbsent(instance.getCompName(),
                k -> {
                  ComponentContainers result = new ComponentContainers();
                  result.setContainers(new ArrayList<>());
                  result.setComponentName(instance.getCompName());
                  return result;
                });

        compContainers.addContainer(instance.getContainerSpec());
      }
    }));
    List<ComponentContainers> result = new ArrayList<>();
    result.addAll(containersByComp.values());
    return result;
  }
@Test
  public void testFilterWithState() throws Exception {
    ServiceContext sc = new MockRunningServiceContext(rule,
        TestServiceManager.createBaseDef("service"));
    GetCompInstancesRequestProto.Builder reqBuilder =
        GetCompInstancesRequestProto.newBuilder();

    reqBuilder.addAllContainerStates(Lists.newArrayList(
        ContainerState.READY.toString()));
    List<ComponentContainers> compContainers = FilterUtils.filterInstances(sc,
        reqBuilder.build());
    Assert.assertEquals("num comps", 2, compContainers.size());
    compContainers.forEach(item -> {
      Assert.assertEquals("num containers", 2, item.getContainers().size());
    });

    reqBuilder.clearContainerStates();
    reqBuilder.addAllContainerStates(Lists.newArrayList(
        ContainerState.STOPPED.toString()));
    Assert.assertEquals("num comps", 0,
        FilterUtils.filterInstances(sc, reqBuilder.build()).size());
  }