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

Code before:
 */
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);
        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();
    }

    private void notifyOfAddedSpanInternal(Span span, io.opentelemetry.api.trace.Span parent) {

        Tracer tracer = tracerProvider.get(span.getScope());
        SpanBuilder spanBuilder = tracer.spanBuilder(span.getName());

        span.getAttributes().forEach(setAttribute(spanBuilder));

        if (parent == null) {
            // root span case
            spanBuilder.setNoParent();
        } else {
            // child / nested span case
            spanBuilder.setParent(Context.current().with(parent));
        }

        io.opentelemetry.api.trace.Span currentOtelSpan =
                spanBuilder
                        .setStartTimestamp(span.getStartTsMillis(), TimeUnit.MILLISECONDS)
                        .startSpan();


Review comment (refactor):
I know this is a back port, I  am wondering if we get an exception processing the spanProcessor shouldn't we still try to clean up spanExporter if we have one. i.e. put the spanExporter code in a finally.  

Code after:
 */
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);
        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() {
        if (spanProcessor != null) {
            spanProcessor.forceFlush();
            spanProcessor.close();
        }
        if (spanExporter != null) {
            spanExporter.flush();
            spanExporter.close();
        }
    }

    private void notifyOfAddedSpanInternal(Span span, io.opentelemetry.api.trace.Span parent) {

        Tracer tracer = tracerProvider.get(span.getScope());
        SpanBuilder spanBuilder = tracer.spanBuilder(span.getName());

        span.getAttributes().forEach(setAttribute(spanBuilder));

        if (parent == null) {
            // root span case
            spanBuilder.setNoParent();
        } else {
            // child / nested span case
            spanBuilder.setParent(Context.current().with(parent));
        }

        io.opentelemetry.api.trace.Span currentOtelSpan =