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

Code before:
    public OpenTelemetryMetricReporter() {
        this(Clock.systemUTC());
    }

    @VisibleForTesting
    OpenTelemetryMetricReporter(Clock clock) {
        this.clock = clock;
    }

    @Override
    public void open(MetricConfig metricConfig) {
        LOG.info("Starting OpenTelemetryMetricReporter");
        super.open(metricConfig);

        final String protocol =
                metricConfig.getProperty(OpenTelemetryReporterOptions.EXPORTER_PROTOCOL.key());

        if (Protocol.HTTP.name().equalsIgnoreCase(protocol)) {
            OtlpHttpMetricExporterBuilder builder = OtlpHttpMetricExporter.builder();
            tryConfigureEndpoint(metricConfig, builder::setEndpoint);
            tryConfigureTimeout(metricConfig, builder::setTimeout);
            exporter = builder.build();
        } else {
            if (!Protocol.gRPC.name().equalsIgnoreCase(protocol)) {
                LOG.warn(
                        "Unknown protocol '{}' for OpenTelemetryEventReporter, defaulting to gRPC",
                        protocol);
            }

            OtlpGrpcMetricExporterBuilder builder = OtlpGrpcMetricExporter.builder();
            tryConfigureEndpoint(metricConfig, builder::setEndpoint);
            tryConfigureTimeout(metricConfig, builder::setTimeout);
            exporter = builder.build();
        }
    }

    @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;

Review comment (suggestion):
```suggestion
                        "Unknown protocol '{}' for OpenTelemetryMetricReporter, defaulting to gRPC",
```

Code after:
        this(Clock.systemUTC());
    }

    @VisibleForTesting
    OpenTelemetryMetricReporter(Clock clock) {
        this.clock = clock;
    }

    @Override
    public void open(MetricConfig metricConfig) {
        LOG.info("Starting OpenTelemetryMetricReporter");
        super.open(metricConfig);

        final String protocol =
                Optional.ofNullable(
                                metricConfig.getProperty(
                                        OpenTelemetryReporterOptions.EXPORTER_PROTOCOL.key()))
                        .orElse("");

        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