public Optional<String> validate(MonitoringInfo monitoringInfo) {

    if (monitoringInfo.getUrn().isEmpty() || monitoringInfo.getType().isEmpty()) {
      return Optional.of(
          String.format(
              "MonitoringInfo requires both urn %s and type %s to be specified.",
              monitoringInfo.getUrn(), monitoringInfo.getType()));
    }

    // Skip checking unknown MonitoringInfos
    Map<String, Set<String>> typeToRequiredLabels = REQUIRED_LABELS.get(monitoringInfo.getUrn());
    if (typeToRequiredLabels == null) {
      return Optional.empty();
    }
    Set<String> requiredLabels = typeToRequiredLabels.get(monitoringInfo.getType());
    if (requiredLabels == null) {
      return Optional.empty();
    }

    // TODO(ajamato): Tighten this restriction to use set equality, to catch unused
    if (!monitoringInfo.getLabelsMap().keySet().containsAll(requiredLabels)) {
      return Optional.of(
          String.format(
              "MonitoringInfo with urn: %s should have labels: %s, actual: %s",
              monitoringInfo.getUrn(), requiredLabels, monitoringInfo.getLabelsMap()));
    }

    return Optional.empty();
  }
@Test
  public void validateReturnsNoErrorOnValidMonitoringInfo() {
    MonitoringInfo testInput =
        MonitoringInfo.newBuilder()
            .setUrn(Urns.USER_SUM_INT64)
            .putLabels(MonitoringInfoConstants.Labels.NAME, "anyCounter")
            .putLabels(MonitoringInfoConstants.Labels.NAMESPACE, "")
            .putLabels(MonitoringInfoConstants.Labels.PTRANSFORM, "anyString")
            .setType(TypeUrns.SUM_INT64_TYPE)
            .build();
    assertFalse(new SpecMonitoringInfoValidator().validate(testInput).isPresent());

    testInput =
        MonitoringInfo.newBuilder()
            .setUrn(Urns.USER_DISTRIBUTION_INT64)
            .putLabels(MonitoringInfoConstants.Labels.NAME, "anyDistribution")
            .putLabels(MonitoringInfoConstants.Labels.NAMESPACE, "namespace")
            .putLabels(MonitoringInfoConstants.Labels.PTRANSFORM, "anyString")
            .setType(TypeUrns.DISTRIBUTION_INT64_TYPE)
            .putLabels("dummy", "value")
            .build();
    assertFalse(new SpecMonitoringInfoValidator().validate(testInput).isPresent());

    testInput =
        MonitoringInfo.newBuilder()
            .setUrn(MonitoringInfoConstants.Urns.ELEMENT_COUNT)
            .setType(TypeUrns.SUM_INT64_TYPE)
            .putLabels(MonitoringInfoConstants.Labels.PTRANSFORM, "value")
            .putLabels(MonitoringInfoConstants.Labels.PCOLLECTION, "anotherValue")
            .build();
    assertFalse(new SpecMonitoringInfoValidator().validate(testInput).isPresent());
  }