Handling 1.0.0 of context-propagation

reactor-core/src/main/java/reactor/core/publisher/ContextPropagation.java

@@ -82,7 +82,10 @@ static <C> ContextSnapshot.Scope setThreadLocals(Object context) {
 				Object value = ((ContextAccessor<C, ?>) contextAccessor).readValue((C) context, key);
 				previousValues = setThreadLocal(key, value, threadLocalAccessor, previousValues);
 			}
-			return ReactorScopeImpl.from(previousValues, registry);
+			if (ContextPropagationSupport.isContextPropagation101Available()) {
+				return ReactorScopeImpl.from(previousValues, registry);
+			}
+			return ReactorScopeImpl100.from(previousValues, registry);
 		}
 	}
 
@@ -491,4 +494,42 @@ public static ContextSnapshot.Scope from(@Nullable Map<Object, Object> previousV
 			});
 		}
 	}
+
+	private static class ReactorScopeImpl100 implements ContextSnapshot.Scope {
+
+		private final Map<Object, Object> previousValues;
+
+		private final ContextRegistry contextRegistry;
+
+		private ReactorScopeImpl100(Map<Object, Object> previousValues,
+				ContextRegistry contextRegistry) {
+			this.previousValues = previousValues;
+			this.contextRegistry = contextRegistry;
+		}
+
+		@Override
+		public void close() {
+			for (ThreadLocalAccessor<?> accessor : this.contextRegistry.getThreadLocalAccessors()) {
+				if (this.previousValues.containsKey(accessor.key())) {
+					Object previousValue = this.previousValues.get(accessor.key());
+					resetThreadLocalValue(accessor, previousValue);
+				}
+			}
+		}
+
+		@SuppressWarnings("unchecked")
+		private <V> void resetThreadLocalValue(ThreadLocalAccessor<?> accessor, @Nullable V previousValue) {
+			if (previousValue != null) {
+				((ThreadLocalAccessor<V>) accessor).setValue(previousValue);
+			}
+			else {
+				accessor.reset();
+			}
+		}
+
+		public static ContextSnapshot.Scope from(@Nullable Map<Object, Object> previousValues, ContextRegistry registry) {
+			return (previousValues != null ? new ReactorScopeImpl100(previousValues, registry) : () -> {
+			});
+		}
+	}
 }

reactor-core/src/main/java/reactor/core/publisher/ContextPropagationSupport.java

@@ -28,24 +28,36 @@ final class ContextPropagationSupport {
     // Ultimately the long term solution should be provided by Reactor Core.
     static final boolean isContextPropagationOnClasspath;
     static final boolean isContextPropagation103OnClasspath;
+    static final boolean isContextPropagation101OnClasspath;
     static boolean       propagateContextToThreadLocals = false;
 
     static {
         boolean contextPropagation = false;
         boolean contextPropagation103 = false;
+        boolean contextPropagation101 = false;
         try {
             Class.forName("io.micrometer.context.ContextRegistry");
             contextPropagation = true;
+            Class.forName("io.micrometer.context.ThreadLocalAccessor").getDeclaredMethod("restore", Object.class);
+            contextPropagation101 = true;
             Class.forName("io.micrometer.context.ContextSnapshotFactory");
             contextPropagation103 = true;
         } catch (ClassNotFoundException notFound) {
+        } catch (NoSuchMethodException notFound) {
         } catch (LinkageError linkageErr) {
         } catch (Throwable err) {
             LOGGER.error("Unexpected exception while detecting ContextPropagation feature." +
                     " The feature is considered disabled due to this:", err);
         }
         isContextPropagationOnClasspath = contextPropagation;
+        isContextPropagation101OnClasspath = contextPropagation101;
         isContextPropagation103OnClasspath = contextPropagation103;
+
+        if (isContextPropagationOnClasspath && !isContextPropagation103OnClasspath) {
+            LOGGER.warn("context-propagation version below 1.0.3 can cause memory leaks" +
+                    " when working with scope-based ThreadLocalAccessors, please " +
+                    "upgrade!");
+        }
     }
 
     /**
@@ -57,6 +69,10 @@ static boolean isContextPropagationAvailable() {
         return isContextPropagationOnClasspath;
     }
 
+    static boolean isContextPropagation101Available() {
+        return isContextPropagation101OnClasspath;
+    }
+
     static boolean isContextPropagation103Available() {
         return isContextPropagation103OnClasspath;
     }