class OnlineServingService implements ServingService {

  @Override
  public GetOnlineFeaturesResponse getOnlineFeatures(GetOnlineFeaturesRequest request) {
    try (Scope scope = tracer.buildSpan("getOnlineFeatures").startActive(true)) {
      List<EntityRow> entityRows = request.getEntityRowsList();
      // Collect the feature/entity value for each entity row in entityValueMap
      Map<EntityRow, Map<String, Value>> entityValuesMap =
          entityRows.stream().collect(Collectors.toMap(row -> row, row -> new HashMap<>()));
      // Collect the feature/entity status metadata for each entity row in entityValueMap
      Map<EntityRow, Map<String, FieldStatus>> entityStatusesMap =
          entityRows.stream().collect(Collectors.toMap(row -> row, row -> new HashMap<>()));
      // Collect featureRows retrieved for logging/tracing
      List<List<Optional<FeatureRow>>> logFeatureRows = new LinkedList<>();

      if (!request.getOmitEntitiesInResponse()) {
        // Add entity row's fields as response fields
        entityRows.forEach(
            entityRow -> {
              Map<String, Value> valueMap = entityRow.getFieldsMap();
              entityValuesMap.get(entityRow).putAll(valueMap);
              entityStatusesMap.get(entityRow).putAll(getMetadataMap(valueMap, false, false));
            });
      }

      List<FeatureSetRequest> featureSetRequests =
          specService.getFeatureSets(request.getFeaturesList(), request.getProject());
      for (FeatureSetRequest featureSetRequest : featureSetRequests) {
        // Pull feature rows for given entity rows from the feature/featureset specified in feature
        // set request.
        // from the configured online
        List<Optional<FeatureRow>> featureRows =
            retriever.getOnlineFeatures(entityRows, featureSetRequest);
        // Check that feature row returned corresponds to a given entity row.
        if (featureRows.size() != entityRows.size()) {
          throw Status.INTERNAL
              .withDescription(
                  "The no. of FeatureRow obtained from OnlineRetriever"
                      + "does not match no. of entityRow passed.")
              .asRuntimeException();
        }

        Streams.zip(entityRows.stream(), featureRows.stream(), Pair::of)
            .forEach(
                entityFeaturePair -> {
                  EntityRow entityRow = entityFeaturePair.getLeft();
                  Optional<FeatureRow> featureRow = entityFeaturePair.getRight();
                  // Unpack feature field values and merge into entityValueMap
                  boolean isOutsideMaxAge =
                      checkOutsideMaxAge(featureSetRequest, entityRow, featureRow);
                  Map<String, Value> valueMap =
                      unpackValueMap(featureRow, featureSetRequest, isOutsideMaxAge);
                  entityValuesMap.get(entityRow).putAll(valueMap);

                  // Generate metadata for feature values and merge into entityFieldsMap
                  boolean isNotFound = featureRow.isEmpty();
                  Map<String, FieldStatus> statusMap =
                      getMetadataMap(valueMap, isNotFound, isOutsideMaxAge);
                  entityStatusesMap.get(entityRow).putAll(statusMap);

                  // Populate metrics/log request
                  populateCountMetrics(statusMap, featureSetRequest);
                });
        populateRequestCountMetrics(featureSetRequest);
        logFeatureRows.add(featureRows);
      }
      if (scope != null) {
        logFeatureRowsTrace(scope, logFeatureRows, featureSetRequests);
      }

      // Build response field values from entityValuesMap and entityStatusesMap
      // Reponse field values should be in the same order as the entityRows provided by the user.
      List<FieldValues> fieldValuesList =
          entityRows.stream()
              .map(
                  entityRow -> {
                    return FieldValues.newBuilder()
                        .putAllFields(entityValuesMap.get(entityRow))
                        .putAllStatuses(entityStatusesMap.get(entityRow))
                        .build();
                  })
              .collect(Collectors.toList());
      return GetOnlineFeaturesResponse.newBuilder().addAllFieldValues(fieldValuesList).build();
    }
  }

  public  OnlineServingService(
      OnlineRetriever retriever, CachedSpecService specService, Tracer tracer);

}