Revert "Review fixes for reactor-3.1:library"

This reverts commit 8c8e72f.

Author: trask

instrumentation/reactor/reactor-3.1/library/src/main/java/io/opentelemetry/instrumentation/reactor/v3_1/ContextPropagationOperator.java

@@ -58,13 +58,8 @@ public final class ContextPropagationOperator {
   @Nullable
   private static final MethodHandle FLUX_CONTEXT_WRITE_METHOD = getContextWriteMethod(Flux.class);
 
-  private static final String SCHEDULERS_HOOK_KEY = RunnableWrapper.class.getName();
-
   @Nullable private static final MethodHandle SCHEDULERS_HOOK_METHOD = getSchedulersHookMethod();
 
-  @Nullable
-  private static final MethodHandle SCHEDULERS_RESET_HOOK_METHOD = getSchedulersResetHookMethod();
-
   @Nullable
   private static MethodHandle getContextWriteMethod(Class<?> type) {
     MethodHandles.Lookup lookup = MethodHandles.publicLookup();
@@ -93,18 +88,6 @@ private static MethodHandle getSchedulersHookMethod() {
     return null;
   }
 
-  @Nullable
-  private static MethodHandle getSchedulersResetHookMethod() {
-    MethodHandles.Lookup lookup = MethodHandles.publicLookup();
-    try {
-      return lookup.findStatic(
-          Schedulers.class, "resetOnScheduleHook", methodType(void.class, String.class));
-    } catch (NoSuchMethodException | IllegalAccessException e) {
-      // ignore
-    }
-    return null;
-  }
-
   public static ContextPropagationOperator create() {
     return builder().build();
   }
@@ -188,7 +171,7 @@ public void registerOnEachOperator() {
       Hooks.onEachOperator(
           TracingSubscriber.class.getName(), tracingLift(asyncOperationEndStrategy));
       AsyncOperationEndStrategies.instance().registerStrategy(asyncOperationEndStrategy);
-      registerScheduleHook(SCHEDULERS_HOOK_KEY, RunnableWrapper::new);
+      registerScheduleHook(RunnableWrapper.class.getName(), RunnableWrapper::new);
       enabled = true;
     }
   }
@@ -204,17 +187,6 @@ private static void registerScheduleHook(String key, Function<Runnable, Runnable
     }
   }
 
-  private static void resetScheduleHook(String key) {
-    if (SCHEDULERS_RESET_HOOK_METHOD == null) {
-      return;
-    }
-    try {
-      SCHEDULERS_RESET_HOOK_METHOD.invoke(key);
-    } catch (Throwable t) {
-      logger.log(WARNING, "Failed to remove scheduler hook", t);
-    }
-  }
-
   /** Unregisters the hook registered by {@link #registerOnEachOperator()}. */
   public void resetOnEachOperator() {
     synchronized (lock) {
@@ -223,7 +195,6 @@ public void resetOnEachOperator() {
       }
       Hooks.resetOnEachOperator(TracingSubscriber.class.getName());
       AsyncOperationEndStrategies.instance().unregisterStrategy(asyncOperationEndStrategy);
-      resetScheduleHook(SCHEDULERS_HOOK_KEY);
       enabled = false;
     }
   }

instrumentation/reactor/reactor-3.1/library/src/test/java/io/opentelemetry/instrumentation/reactor/v3_1/HooksTest.java

@@ -6,19 +6,10 @@
 package io.opentelemetry.instrumentation.reactor.v3_1;
 
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
-import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.assertj.core.api.Assertions.assertThat;
 
-import io.opentelemetry.api.trace.Span;
-import io.opentelemetry.api.trace.SpanContext;
-import io.opentelemetry.api.trace.TraceFlags;
-import io.opentelemetry.api.trace.TraceState;
-import io.opentelemetry.context.Context;
-import io.opentelemetry.context.Scope;
 import java.util.concurrent.Callable;
-import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.function.Function;
 import org.junit.jupiter.api.Test;
 import reactor.core.CoreSubscriber;
 import reactor.core.Disposable;
@@ -27,14 +18,6 @@
 
 class HooksTest {
 
-  private static final Span PARENT_SPAN =
-      Span.wrap(
-          SpanContext.create(
-              "11111111111111111111111111111111",
-              "1111111111111111",
-              TraceFlags.getSampled(),
-              TraceState.getDefault()));
-
   @Test
   void canResetOurHooks() {
     ContextPropagationOperator operator = ContextPropagationOperator.create();
@@ -52,25 +35,6 @@ void canResetOurHooks() {
     assertThat(subscriber.get()).extracting("actual").isNotInstanceOf(TracingSubscriber.class);
   }
 
-  @Test
-  void canResetSchedulerHook() throws InterruptedException {
-    if (!schedulerHooksSupported()) {
-      return;
-    }
-
-    ContextPropagationOperator operator = ContextPropagationOperator.create();
-
-    assertThat(scheduledSpanIsCurrent()).isFalse();
-
-    operator.registerOnEachOperator();
-    try {
-      assertThat(scheduledSpanIsCurrent()).isTrue();
-    } finally {
-      operator.resetOnEachOperator();
-    }
-    assertThat(scheduledSpanIsCurrent()).isFalse();
-  }
-
   private static class CapturingMono extends Mono<Integer> {
     private final AtomicReference<CoreSubscriber<? super Integer>> subscriber;
 
@@ -107,29 +71,4 @@ void testInvalidBlockUsage() throws InterruptedException {
     disposable.dispose();
     operator.resetOnEachOperator();
   }
-
-  private static boolean schedulerHooksSupported() {
-    try {
-      Schedulers.class.getMethod("onScheduleHook", String.class, Function.class);
-      Schedulers.class.getMethod("resetOnScheduleHook", String.class);
-      return true;
-    } catch (NoSuchMethodException e) {
-      return false;
-    }
-  }
-
-  private static boolean scheduledSpanIsCurrent() throws InterruptedException {
-    CountDownLatch latch = new CountDownLatch(1);
-    AtomicReference<Boolean> currentSpanValid = new AtomicReference<>(false);
-    try (Scope ignored = Context.root().with(PARENT_SPAN).makeCurrent()) {
-      Schedulers.single()
-          .schedule(
-              () -> {
-                currentSpanValid.set(Span.current().getSpanContext().isValid());
-                latch.countDown();
-              });
-    }
-    assertThat(latch.await(5, SECONDS)).isTrue();
-    return currentSpanValid.get();
-  }
 }