Repository: apache/flink
File: flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/OpenTelemetryMetricReporter.java (Java)

Code before:
        switch (protocol.toLowerCase()) {
            case "http":
                OtlpHttpMetricExporterBuilder httpBuilder = OtlpHttpMetricExporter.builder();
                tryConfigureEndpoint(metricConfig, httpBuilder::setEndpoint);
                tryConfigureTimeout(metricConfig, httpBuilder::setTimeout);
                exporter = httpBuilder.build();
                break;
            default:
                LOG.warn(
                        "Unknown protocol '{}' for OpenTelemetryMetricReporter, defaulting to gRPC",
                        protocol);
            // Fall through to the "gRPC" case
            case "grpc":
                OtlpGrpcMetricExporterBuilder grpcBuilder = OtlpGrpcMetricExporter.builder();
                tryConfigureEndpoint(metricConfig, grpcBuilder::setEndpoint);
                tryConfigureTimeout(metricConfig, grpcBuilder::setTimeout);
                exporter = grpcBuilder.build();
                break;
        }
    }

    @Override
    public void close() {
        exporter.flush();
        waitForLastReportToComplete();
        exporter.close();
    }

    @Override
    public void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group) {
        final String name =
                LOGICAL_SCOPE_PREFIX
                        + LogicalScopeProvider.castFrom(group)
                                .getLogicalScope(CharacterFilter.NO_OP_FILTER)
                        + "."
                        + metricName;

        Map<String, String> variables =
                group.getAllVariables().entrySet().stream()
                        .collect(
                                Collectors.toMap(
                                        e -> VariableNameUtil.getVariableName(e.getKey()),
                                        Entry::getValue));
        LOG.debug("Adding metric {} with variables {}", metricName, variables);

        MetricMetadata metricMetadata = new MetricMetadata(name, variables);

        synchronized (this) {
            switch (metric.getMetricType()) {
                case COUNTER:
                    this.counters.put((Counter) metric, metricMetadata);

Review comment (suggestion):
I notice that  `waitForLastReportToComplete` is looking at lastResult, I wonder if we should check lastResult,is not null then issuing waitForLastReportToComplete(); when exporter is null. So that we would wait for it to finish in the close if there is ongoing work. WDYT?

Looks good otherwise.

Code after:
        switch (protocol.toLowerCase()) {
            case "http":
                OtlpHttpMetricExporterBuilder httpBuilder = OtlpHttpMetricExporter.builder();
                tryConfigureEndpoint(metricConfig, httpBuilder::setEndpoint);
                tryConfigureTimeout(metricConfig, httpBuilder::setTimeout);
                exporter = httpBuilder.build();
                break;
            default:
                LOG.warn(
                        "Unknown protocol '{}' for OpenTelemetryMetricReporter, defaulting to gRPC",
                        protocol);
            // Fall through to the "gRPC" case
            case "grpc":
                OtlpGrpcMetricExporterBuilder grpcBuilder = OtlpGrpcMetricExporter.builder();
                tryConfigureEndpoint(metricConfig, grpcBuilder::setEndpoint);
                tryConfigureTimeout(metricConfig, grpcBuilder::setTimeout);
                exporter = grpcBuilder.build();
                break;
        }
    }

    @Override
    public void close() {
        if (exporter != null) {
            exporter.flush();
            waitForLastReportToComplete();
            exporter.close();
        }
    }

    @Override
    public void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group) {
        final String name =
                LOGICAL_SCOPE_PREFIX
                        + LogicalScopeProvider.castFrom(group)
                                .getLogicalScope(CharacterFilter.NO_OP_FILTER)
                        + "."
                        + metricName;

        Map<String, String> variables =
                group.getAllVariables().entrySet().stream()
                        .collect(
                                Collectors.toMap(
                                        e -> VariableNameUtil.getVariableName(e.getKey()),
                                        Entry::getValue));
        LOG.debug("Adding metric {} with variables {}", metricName, variables);

        MetricMetadata metricMetadata = new MetricMetadata(name, variables);

        synchronized (this) {
            switch (metric.getMetricType()) {