[improve][misc] Follow up on ObjectMapper sharing changes (#19228)

buildtools/src/main/java/org/apache/pulsar/tests/SingletonCleanerListener.java

@@ -30,8 +30,8 @@
  */
 public class SingletonCleanerListener extends BetweenTestClassesListenerAdapter {
     private static final Logger LOG = LoggerFactory.getLogger(SingletonCleanerListener.class);
-    private static final Method OBJECTMAPPERFACTORY_CLEANCACHES_METHOD;
-    private static final Method OBJECTMAPPERFACTORY_REFRESH_METHOD;
+    private static final Method OBJECTMAPPERFACTORY_CLEARCACHES_METHOD;
+    private static final Method JSONSCHEMA_CLEARCACHES_METHOD;
 
     static {
         Class<?> objectMapperFactoryClazz =
@@ -50,49 +50,57 @@ public class SingletonCleanerListener extends BetweenTestClassesListenerAdapter
                                 .getMethod("clearCaches");
             }
         } catch (NoSuchMethodException e) {
-            LOG.warn("Cannot find method for cleaning singleton ObjectMapper caches", e);
+            LOG.warn("Cannot find method for clearing singleton ObjectMapper caches", e);
         }
-        OBJECTMAPPERFACTORY_CLEANCACHES_METHOD = clearCachesMethod;
+        OBJECTMAPPERFACTORY_CLEARCACHES_METHOD = clearCachesMethod;
 
-        Method refreshMethod = null;
+
+        Class<?> jsonSchemaClazz = null;
         try {
-            if (objectMapperFactoryClazz != null) {
-                refreshMethod =
-                        objectMapperFactoryClazz
-                                .getMethod("refresh");
+            jsonSchemaClazz = ClassUtils.getClass("org.apache.pulsar.client.impl.schema.JSONSchema");
+        } catch (ClassNotFoundException e) {
+            LOG.warn("Cannot find JSONSchema class", e);
+        }
+
+        Method jsonSchemaCleanCachesMethod = null;
+        try {
+            if (jsonSchemaClazz != null) {
+                jsonSchemaCleanCachesMethod =
+                        jsonSchemaClazz
+                                .getMethod("clearCaches");
             }
         } catch (NoSuchMethodException e) {
-            LOG.warn("Cannot find method for refreshing singleton ObjectMapper instances", e);
+            LOG.warn("Cannot find method for clearing singleton JSONSchema caches", e);
         }
-        OBJECTMAPPERFACTORY_REFRESH_METHOD = refreshMethod;
+        JSONSCHEMA_CLEARCACHES_METHOD = jsonSchemaCleanCachesMethod;
     }
 
     @Override
     protected void onBetweenTestClasses(Class<?> endedTestClass, Class<?> startedTestClass) {
-        cleanObjectMapperFactoryCaches();
-        refreshObjectMapperFactory();
+        objectMapperFactoryClearCaches();
+        jsonSchemaClearCaches();
     }
 
     // Call ObjectMapperFactory.clearCaches() using reflection to clear up classes held in
     // the singleton Jackson ObjectMapper instances
-    private static void cleanObjectMapperFactoryCaches() {
-        if (OBJECTMAPPERFACTORY_CLEANCACHES_METHOD != null) {
+    private static void objectMapperFactoryClearCaches() {
+        if (OBJECTMAPPERFACTORY_CLEARCACHES_METHOD != null) {
             try {
-                OBJECTMAPPERFACTORY_CLEANCACHES_METHOD.invoke(null);
+                OBJECTMAPPERFACTORY_CLEARCACHES_METHOD.invoke(null);
             } catch (IllegalAccessException | InvocationTargetException e) {
                 LOG.warn("Cannot clean singleton ObjectMapper caches", e);
             }
         }
     }
 
-    // Call ObjectMapperFactory.refresh() using reflection to release ObjectMapper instances
-    // that might be holding on classloaders and classes
-    private static void refreshObjectMapperFactory() {
-        if (OBJECTMAPPERFACTORY_REFRESH_METHOD != null) {
+    // Call JSONSchema.clearCaches() using reflection to clear up classes held in
+    // the singleton Jackson ObjectMapper instance of JSONSchema class
+    private static void jsonSchemaClearCaches() {
+        if (JSONSCHEMA_CLEARCACHES_METHOD != null) {
             try {
-                OBJECTMAPPERFACTORY_REFRESH_METHOD.invoke(null);
+                JSONSCHEMA_CLEARCACHES_METHOD.invoke(null);
             } catch (IllegalAccessException | InvocationTargetException e) {
-                LOG.warn("Cannot refresh ObjectMapper instances", e);
+                LOG.warn("Cannot clean singleton JSONSchema caches", e);
             }
         }
     }

pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/JSONSchema.java

@@ -27,6 +27,7 @@
 import com.fasterxml.jackson.module.jsonSchema.JsonSchema;
 import com.fasterxml.jackson.module.jsonSchema.JsonSchemaGenerator;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.client.api.schema.SchemaDefinition;
 import org.apache.pulsar.client.api.schema.SchemaReader;
@@ -42,15 +43,21 @@
  */
 @Slf4j
 public class JSONSchema<T> extends AvroBaseStructSchema<T> {
-    private static final ObjectMapper JSON_MAPPER = createObjectMapper();
+    private static final AtomicReference<ObjectMapper> JSON_MAPPER = new AtomicReference<>(createObjectMapper());
 
     private static ObjectMapper createObjectMapper() {
-        ObjectMapper mapper = new ObjectMapper();
+        ObjectMapper mapper = ObjectMapperFactory.create();
         mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+        // keep backwards compatibility, don't accept unknown enum values
+        mapper.configure(DeserializationFeature.READ_UNKNOWN_ENUM_VALUES_AS_NULL, false);
         mapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
         return mapper;
     }
 
+    private static ObjectMapper jsonMapper() {
+        return JSON_MAPPER.get();
+    }
+
     private final Class<T> pojo;
 
     private JSONSchema(SchemaInfo schemaInfo, Class<T> pojo, SchemaReader<T> reader, SchemaWriter<T> writer) {
@@ -87,9 +94,9 @@ public SchemaInfo getBackwardsCompatibleJsonSchemaInfo() {
 
     public static <T> JSONSchema<T> of(SchemaDefinition<T> schemaDefinition) {
         SchemaReader<T> reader = schemaDefinition.getSchemaReaderOpt()
-                .orElseGet(() -> new JacksonJsonReader<>(JSON_MAPPER, schemaDefinition.getPojo()));
+                .orElseGet(() -> new JacksonJsonReader<>(jsonMapper(), schemaDefinition.getPojo()));
         SchemaWriter<T> writer = schemaDefinition.getSchemaWriterOpt()
-                .orElseGet(() -> new JacksonJsonWriter<>(JSON_MAPPER));
+                .orElseGet(() -> new JacksonJsonWriter<>(jsonMapper()));
         return new JSONSchema<>(parseSchemaInfo(schemaDefinition, SchemaType.JSON), schemaDefinition.getPojo(),
                 reader, writer);
     }
@@ -102,4 +109,18 @@ public static <T> JSONSchema<T> of(Class<T> pojo, Map<String, String> properties
         return JSONSchema.of(SchemaDefinition.<T>builder().withPojo(pojo).withProperties(properties).build());
     }
 
+    /**
+     * Clears the caches tied to the ObjectMapper instances and replaces the singleton ObjectMapper instance.
+     *
+     * This can be used in tests to ensure that classloaders and class references don't leak across tests.
+     */
+    public static void clearCaches() {
+        jsonMapper().getTypeFactory().clearCache();
+        replaceSingletonInstance();
+    }
+
+    private static void replaceSingletonInstance() {
+        // recycle the singleton instance to release remaining caches
+        JSON_MAPPER.set(createObjectMapper());
+    }
 }

pulsar-common/src/main/java/org/apache/pulsar/common/util/ObjectMapperFactory.java

@@ -266,24 +266,26 @@ private static void setAnnotationsModule(ObjectMapper mapper) {
     }
 
     /**
-     * Clears the caches tied to the ObjectMapper instances.
-     * This is used in tests to ensure that classloaders and class references don't leak between tests.
-     * Jackson's ObjectMapper doesn't expose a public API for clearing all caches so this solution is partial.
+     * Clears the caches tied to the ObjectMapper instances and replaces the singleton ObjectMapper instance.
+     *
+     * This can be used in tests to ensure that classloaders and class references don't leak across tests.
      */
     public static void clearCaches() {
-        clearCachesForObjectMapper(getMapper().getObjectMapper());
-        clearCachesForObjectMapper(getYamlMapper().getObjectMapper());
+        clearTypeFactoryCache(getMapper().getObjectMapper());
+        clearTypeFactoryCache(getYamlMapper().getObjectMapper());
+        clearTypeFactoryCache(getMapperWithIncludeAlways().getObjectMapper());
+        replaceSingletonInstances();
     }
 
-    private static void clearCachesForObjectMapper(ObjectMapper objectMapper) {
+    private static void clearTypeFactoryCache(ObjectMapper objectMapper) {
         objectMapper.getTypeFactory().clearCache();
     }
 
-    /**
+    /*
      * Replaces the existing singleton ObjectMapper instances with new instances.
      * This is used in tests to ensure that classloaders and class references don't leak between tests.
      */
-    public static void refresh() {
+    private static void replaceSingletonInstances() {
         MAPPER_REFERENCE.set(new MapperReference(createObjectMapperInstance()));
         INSTANCE_WITH_INCLUDE_ALWAYS.set(new MapperReference(createObjectMapperWithIncludeAlways()));
         YAML_MAPPER_REFERENCE.set(new MapperReference(createYamlInstance()));