@Override
  public Optional<ComputationConfig> fetchConfig(String computationId) {
    Preconditions.checkArgument(
        !computationId.isEmpty(),
        "computationId is empty. Cannot fetch computation config without a computationId.");

    GetConfigResponse response =
        applianceComputationConfigFetcher.fetchConfig(
            GetConfigRequest.newBuilder().addComputations(computationId).build());

    if (response == null) {
      return Optional.empty();
    }

    for (Windmill.GetConfigResponse.SystemNameToComputationIdMapEntry entry :
        response.getSystemNameToComputationIdMapList()) {
      systemNameToComputationIdMap.put(entry.getSystemName(), entry.getComputationId());
    }

    return createComputationConfig(
        // We are only fetching the config for 1 computation, so we should only be getting that
        // computation back.
        Iterables.getOnlyElement(response.getCloudWorksList()),
        transformUserNameToStateFamilyByComputationId(response),
        response.getNameMapList().stream()
            .collect(toImmutableMap(NameMapEntry::getUserName, NameMapEntry::getSystemName)));
  }
@Test
  public void testGetComputationConfig() throws IOException {
    List<Windmill.GetConfigResponse.NameMapEntry> nameMapEntries =
        ImmutableList.of(
            Windmill.GetConfigResponse.NameMapEntry.newBuilder()
                .setUserName("userName1")
                .setSystemName("systemName1")
                .build(),
            Windmill.GetConfigResponse.NameMapEntry.newBuilder()
                .setUserName("userName2")
                .setSystemName("userName2")
                .build());
    String serializedMapTask =
        Transport.getJsonFactory()
            .toString(
                new MapTask()
                    .setSystemName("systemName")
                    .setStageName("stageName")
                    .setInstructions(ImmutableList.of()));
    Windmill.GetConfigResponse getConfigResponse =
        Windmill.GetConfigResponse.newBuilder()
            .addAllNameMap(nameMapEntries)
            .addComputationConfigMap(
                Windmill.GetConfigResponse.ComputationConfigMapEntry.newBuilder()
                    .setComputationId("systemName")
                    .setComputationConfig(
                        Windmill.ComputationConfig.newBuilder()
                            .addTransformUserNameToStateFamily(
                                Windmill.ComputationConfig.TransformUserNameToStateFamilyEntry
                                    .newBuilder()
                                    .setStateFamily("stateFamilyName")
                                    .setTransformUserName("transformUserName")
                                    .build())
                            .build())
                    .build())
            .addCloudWorks(serializedMapTask)
            .build();
    ComputationConfig expectedConfig =
        ComputationConfig.create(
            Transport.getJsonFactory().fromString(serializedMapTask, MapTask.class),
            getConfigResponse.getComputationConfigMapList().stream()
                .map(Windmill.GetConfigResponse.ComputationConfigMapEntry::getComputationConfig)
                .flatMap(
                    computationConfig ->
                        computationConfig.getTransformUserNameToStateFamilyList().stream())
                .collect(
                    toMap(
                        Windmill.ComputationConfig.TransformUserNameToStateFamilyEntry
                            ::getTransformUserName,
                        Windmill.ComputationConfig.TransformUserNameToStateFamilyEntry
                            ::getStateFamily)),
            nameMapEntries.stream()
                .collect(
                    toMap(
                        Windmill.GetConfigResponse.NameMapEntry::getUserName,
                        Windmill.GetConfigResponse.NameMapEntry::getSystemName)));
    StreamingApplianceComputationConfigFetcher configLoader =
        createStreamingApplianceConfigLoader();
    when(mockWindmillServer.getConfig(any())).thenReturn(getConfigResponse);
    Optional<ComputationConfig> config = configLoader.fetchConfig("someComputationId");
    assertTrue(config.isPresent());
    assertThat(config.get()).isEqualTo(expectedConfig);
  }