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

Code before:
 * A Flink {@link org.apache.flink.traces.reporter.TraceReporter} which is made to export spans
 * using Open Telemetry's {@link SpanExporter}.
 */
public class OpenTelemetryTraceReporter extends OpenTelemetryReporterBase implements TraceReporter {
    private static final Logger LOG = LoggerFactory.getLogger(OpenTelemetryTraceReporter.class);
    private SpanExporter spanExporter;
    private TracerProvider tracerProvider;
    private BatchSpanProcessor spanProcessor;

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

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

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

            OtlpGrpcSpanExporterBuilder builder = OtlpGrpcSpanExporter.builder();
            tryConfigureEndpoint(metricConfig, builder::setEndpoint);
            tryConfigureTimeout(metricConfig, builder::setTimeout);
            spanExporter = builder.build();
        }

        spanProcessor = BatchSpanProcessor.builder(spanExporter).build();
        tracerProvider =
                SdkTracerProvider.builder()
                        .addSpanProcessor(spanProcessor)
                        .setResource(resource)
                        .build();
    }

    @Override
    public void close() {
        spanProcessor.forceFlush();
        spanProcessor.close();
        spanExporter.flush();
        spanExporter.close();
    }


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

Code after:
 * A Flink {@link org.apache.flink.traces.reporter.TraceReporter} which is made to export spans
 * using Open Telemetry's {@link SpanExporter}.
 */
public class OpenTelemetryTraceReporter extends OpenTelemetryReporterBase implements TraceReporter {
    private static final Logger LOG = LoggerFactory.getLogger(OpenTelemetryTraceReporter.class);
    private SpanExporter spanExporter;
    private TracerProvider tracerProvider;
    private BatchSpanProcessor spanProcessor;

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

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

        switch (protocol.toLowerCase()) {
            case "http":
                OtlpHttpSpanExporterBuilder httpBuilder = OtlpHttpSpanExporter.builder();
                tryConfigureEndpoint(metricConfig, httpBuilder::setEndpoint);
                tryConfigureTimeout(metricConfig, httpBuilder::setTimeout);
                spanExporter = httpBuilder.build();
                break;
            default:
                LOG.warn(
                        "Unknown protocol '{}' for OpenTelemetryTraceReporter, defaulting to gRPC",
                        protocol);
            // Fall through to the "gRPC" case
            case "grpc":
                OtlpGrpcSpanExporterBuilder grpcBuilder = OtlpGrpcSpanExporter.builder();
                tryConfigureEndpoint(metricConfig, grpcBuilder::setEndpoint);
                tryConfigureTimeout(metricConfig, grpcBuilder::setTimeout);
                spanExporter = grpcBuilder.build();
                break;
        }

        spanProcessor = BatchSpanProcessor.builder(spanExporter).build();
        tracerProvider =
                SdkTracerProvider.builder()
                        .addSpanProcessor(spanProcessor)
                        .setResource(resource)
                        .build();
    }

    @Override
    public void close() {
        spanProcessor.forceFlush();