[common][server][controller][pulsar][test] Fail fast (config-gated) when a pub-sub adapter factory class is not configured

clients/da-vinci-client/src/main/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerImpl.java

@@ -220,6 +220,10 @@ public VeniceChangelogConsumerImpl(
       this.storeName =
           VeniceView.getViewStoreName(changelogClientConfig.getStoreName(), changelogClientConfig.getViewName());
       Properties rocksDBBufferProperties = new Properties();
+      // VeniceServerConfig below eagerly builds a PubSubClientsFactory, which fails fast when the pub-sub
+      // adapter factory classes are not configured. Seed the buffer config with the consumer properties (which
+      // carry those classes) so it resolves them; the RocksDB buffer itself does not use a pub-sub client.
+      rocksDBBufferProperties.putAll(changelogClientConfig.getConsumerProperties());
       String rocksDBBufferPath = changelogClientConfig.getBootstrapFileSystemPath();
       if (rocksDBBufferPath == null || rocksDBBufferPath.isEmpty()) {
         throw new VeniceException("bootstrapFileSystemPath must be configured for consuming view store: " + storeName);

clients/da-vinci-client/src/test/java/com/linkedin/davinci/DaVinciBackendTest.java

@@ -47,6 +47,7 @@
 import com.linkedin.venice.schema.writecompute.DerivedSchemaEntry;
 import com.linkedin.venice.serialization.avro.SchemaPresenceChecker;
 import com.linkedin.venice.service.ICProvider;
+import com.linkedin.venice.utils.TestUtils;
 import com.linkedin.venice.utils.VeniceProperties;
 import io.tehuti.metrics.MetricsRepository;
 import java.util.Optional;
@@ -83,6 +84,7 @@ public void setUp() throws Exception {
     serverProps.setProperty(INGESTION_USE_DA_VINCI_CLIENT, "true");
     serverProps.setProperty(DATA_BASE_PATH, "/tmp/test");
     serverProps.setProperty(ROCKSDB_BLOCK_CACHE_SIZE_IN_BYTES, "0");
+    serverProps.putAll(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs());
     VeniceProperties veniceProperties = new VeniceProperties(serverProps);
     VeniceConfigLoader configLoader = new VeniceConfigLoader(veniceProperties);
 

clients/da-vinci-client/src/test/java/com/linkedin/davinci/StoreBackendTest.java

@@ -84,7 +84,8 @@ public class StoreBackendTest {
   @BeforeMethod
   void setUp() {
     baseDataPath = Utils.getTempDataDirectory();
-    VeniceProperties backendConfig = new PropertyBuilder().put(ConfigKeys.CLUSTER_NAME, "test-cluster")
+    VeniceProperties backendConfig = new PropertyBuilder().put(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs())
+        .put(ConfigKeys.CLUSTER_NAME, "test-cluster")
         .put(ConfigKeys.ZOOKEEPER_ADDRESS, "test-zookeeper")
         .put(ConfigKeys.KAFKA_BOOTSTRAP_SERVERS, "test-kafka")
         .put(ConfigKeys.DATA_BASE_PATH, baseDataPath.getAbsolutePath())
@@ -668,7 +669,8 @@ public void testResumePausedSITOnTargetPromotion() throws Exception {
   @Test
   public void testLegacyNonTargetRegionSubscribesOnOnline() throws Exception {
     // Re-create storeBackend with paused-SIT disabled (legacy mode).
-    VeniceProperties legacyConfig = new PropertyBuilder().put(ConfigKeys.CLUSTER_NAME, "test-cluster")
+    VeniceProperties legacyConfig = new PropertyBuilder().put(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs())
+        .put(ConfigKeys.CLUSTER_NAME, "test-cluster")
         .put(ConfigKeys.ZOOKEEPER_ADDRESS, "test-zookeeper")
         .put(ConfigKeys.KAFKA_BOOTSTRAP_SERVERS, "test-kafka")
         .put(ConfigKeys.DATA_BASE_PATH, baseDataPath.getAbsolutePath())
@@ -716,7 +718,8 @@ public void testLegacyNonTargetRegionSubscribesOnOnline() throws Exception {
    * region stays {@code dc-0} and paused-SIT stays enabled.
    */
   private void rebuildStoreBackendWithRollForwardOrder(String rollForwardOrder) {
-    VeniceProperties config = new PropertyBuilder().put(ConfigKeys.CLUSTER_NAME, "test-cluster")
+    VeniceProperties config = new PropertyBuilder().put(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs())
+        .put(ConfigKeys.CLUSTER_NAME, "test-cluster")
         .put(ConfigKeys.ZOOKEEPER_ADDRESS, "test-zookeeper")
         .put(ConfigKeys.KAFKA_BOOTSTRAP_SERVERS, "test-kafka")
         .put(ConfigKeys.DATA_BASE_PATH, baseDataPath.getAbsolutePath())

clients/da-vinci-client/src/test/java/com/linkedin/davinci/VersionBackendTest.java

@@ -167,6 +167,7 @@ public void testRecordTransformerSubscribe() {
         .put(ConfigKeys.KAFKA_BOOTSTRAP_SERVERS, "test-kafka")
         .put(ConfigKeys.DATA_BASE_PATH, baseDataPath.getAbsolutePath())
         .put(ConfigKeys.LOCAL_REGION_NAME, "dc-0")
+        .put(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs())
         .build();
     VeniceConfigLoader veniceConfigLoader = new VeniceConfigLoader(backendConfig);
     when(mockDaVinciBackend.getConfigLoader()).thenReturn(veniceConfigLoader);
@@ -259,6 +260,7 @@ public void testPushStatusDisabledForVersionSpecificClient() {
         .put(ConfigKeys.DATA_BASE_PATH, baseDataPath.getAbsolutePath())
         .put(ConfigKeys.LOCAL_REGION_NAME, "dc-0")
         .put(ConfigKeys.PUSH_STATUS_STORE_ENABLED, true)
+        .put(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs())
         .build();
     VeniceConfigLoader veniceConfigLoader = new VeniceConfigLoader(backendConfig);
     when(mockDaVinciBackend.getConfigLoader()).thenReturn(veniceConfigLoader);

clients/da-vinci-client/src/test/java/com/linkedin/davinci/client/AvroGenericDaVinciClientTest.java

@@ -46,6 +46,7 @@
 import com.linkedin.venice.utils.DaemonThreadFactory;
 import com.linkedin.venice.utils.PropertyBuilder;
 import com.linkedin.venice.utils.ReferenceCounted;
+import com.linkedin.venice.utils.TestUtils;
 import com.linkedin.venice.utils.VeniceProperties;
 import java.lang.reflect.Field;
 import java.security.AccessController;
@@ -85,6 +86,7 @@ public AvroGenericDaVinciClient setUpSpecificClient(ClientConfig clientConfig, b
     DaVinciConfig daVinciConfig = new DaVinciConfig();
     VeniceProperties backendConfig = new PropertyBuilder().put(SERVER_DATABASE_CHECKSUM_VERIFICATION_ENABLED, false)
         .put(DAVINCI_VALIDATE_SPECIFIC_SCHEMA_ENABLED, validateSpecificSchema)
+        .put(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs())
         .build();
 
     AvroGenericDaVinciClient<Integer, String> dvcClient =
@@ -130,6 +132,7 @@ public AvroGenericSeekableDaVinciClient setUpSeekableClient(ClientConfig clientC
     DaVinciConfig daVinciConfig = new DaVinciConfig();
     VeniceProperties backendConfig = new PropertyBuilder().put(SERVER_DATABASE_CHECKSUM_VERIFICATION_ENABLED, false)
         .put(DAVINCI_VALIDATE_SPECIFIC_SCHEMA_ENABLED, validateSpecificSchema)
+        .put(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs())
         .build();
 
     AvroGenericSeekableDaVinciClient<Integer, String> dvcClient = spy(
@@ -210,6 +213,7 @@ public AvroGenericDaVinciClient setUpClientWithRecordTransformer(
 
     VeniceProperties backendConfig =
         new PropertyBuilder().put(SERVER_DATABASE_CHECKSUM_VERIFICATION_ENABLED, enableDatabaseChecksumVerification)
+            .put(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs())
             .build();
 
     AvroGenericDaVinciClient<Integer, String> dvcClient =

clients/da-vinci-client/src/test/java/com/linkedin/davinci/config/VeniceServerConfigTest.java

@@ -28,6 +28,7 @@
 import static org.testng.Assert.assertTrue;
 
 import com.linkedin.davinci.blobtransfer.client.NettyFileTransferClient;
+import com.linkedin.venice.utils.TestUtils;
 import com.linkedin.venice.utils.VeniceProperties;
 import java.util.Arrays;
 import java.util.HashMap;
@@ -42,6 +43,7 @@
 public class VeniceServerConfigTest {
   private Properties populatedBasicProperties() {
     Properties props = new Properties();
+    props.putAll(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs());
     props.setProperty(CLUSTER_NAME, "test_cluster");
     props.setProperty(ZOOKEEPER_ADDRESS, "fake_zk_addr");
     props.setProperty(KAFKA_BOOTSTRAP_SERVERS, "fake_kafka_addr");

clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/ChangelogClientConfigTest.java

@@ -1,8 +1,10 @@
 package com.linkedin.davinci.consumer;
 
+import com.linkedin.venice.utils.TestUtils;
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.BrokenBarrierException;
 import java.util.concurrent.CyclicBarrier;
@@ -33,6 +35,13 @@
 public class ChangelogClientConfigTest {
   private static final String GLOBAL_STORE = "global_store";
 
+  @SuppressWarnings("rawtypes")
+  private static ChangelogClientConfig newGlobalConfig() {
+    Properties consumerProperties = new Properties();
+    consumerProperties.putAll(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs());
+    return new ChangelogClientConfig(GLOBAL_STORE).setConsumerProperties(consumerProperties);
+  }
+
   /**
    * Deterministic (single-threaded) regression test.
    *
@@ -47,7 +56,7 @@ public class ChangelogClientConfigTest {
   @Test
   @SuppressWarnings("rawtypes")
   public void testCloneConfigDoesNotShareInnerClientConfig() {
-    ChangelogClientConfig global = new ChangelogClientConfig(GLOBAL_STORE);
+    ChangelogClientConfig global = newGlobalConfig();
 
     ChangelogClientConfig clone1 = ChangelogClientConfig.cloneConfig(global).setStoreName("store_A");
     ChangelogClientConfig clone2 = ChangelogClientConfig.cloneConfig(global).setStoreName("store_B");
@@ -68,7 +77,7 @@ public void testCloneConfigDoesNotShareInnerClientConfig() {
 
   @Test
   public void testBackgroundReporterThreadSleepIntervalCloned() {
-    ChangelogClientConfig global = new ChangelogClientConfig(GLOBAL_STORE);
+    ChangelogClientConfig global = newGlobalConfig();
 
     // Default should be 60
     Assert.assertEquals(global.getBackgroundReporterThreadSleepIntervalInSeconds(), 60L);
@@ -109,7 +118,7 @@ public void testBackgroundReporterThreadSleepIntervalCloned() {
   public void testCloneConfigIsThreadSafe() throws InterruptedException {
     final int numThreads = 20;
 
-    ChangelogClientConfig global = new ChangelogClientConfig(GLOBAL_STORE);
+    ChangelogClientConfig global = newGlobalConfig();
 
     ChangelogClientConfig[] clones = new ChangelogClientConfig[numThreads];
     String[] expectedNames = new String[numThreads];

clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerClientFactoryTest.java

@@ -31,6 +31,7 @@
 import com.linkedin.venice.pubsub.api.PubSubMessageDeserializer;
 import com.linkedin.venice.schema.SchemaReader;
 import com.linkedin.venice.utils.ObjectMapperFactory;
+import com.linkedin.venice.utils.TestUtils;
 import com.linkedin.venice.views.MaterializedView;
 import io.tehuti.metrics.MetricsRepository;
 import java.nio.charset.StandardCharsets;
@@ -57,6 +58,7 @@ public class VeniceChangelogConsumerClientFactoryTest {
   @Test
   public void testGetChangelogConsumer() throws ExecutionException, InterruptedException, JsonProcessingException {
     Properties consumerProperties = new Properties();
+    consumerProperties.putAll(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs());
     String localKafkaUrl = "http://www.fooAddress.linkedin.com:16337";
     consumerProperties.put(ConfigKeys.PUBSUB_BROKER_ADDRESS, localKafkaUrl);
     consumerProperties.put(ConfigKeys.KAFKA_BOOTSTRAP_SERVERS, localKafkaUrl);
@@ -131,6 +133,7 @@ public void testGetChangelogConsumer() throws ExecutionException, InterruptedExc
   public void testGetChangelogConsumerWithConsumerId()
       throws ExecutionException, InterruptedException, JsonProcessingException {
     Properties consumerProperties = new Properties();
+    consumerProperties.putAll(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs());
     String localKafkaUrl = "http://www.fooAddress.linkedin.com:16337";
     consumerProperties.put(ConfigKeys.PUBSUB_BROKER_ADDRESS, localKafkaUrl);
     SchemaReader mockSchemaReader = Mockito.mock(SchemaReader.class);
@@ -199,6 +202,7 @@ private void setUpMockStoreResponse(D2ControllerClient mockControllerClient, Str
   @Test
   public void testGetChangelogConsumerThrowsException() {
     Properties consumerProperties = new Properties();
+    consumerProperties.putAll(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs());
     String localKafkaUrl = "http://www.fooAddress.linkedin.com:16337";
     consumerProperties.put(ConfigKeys.PUBSUB_BROKER_ADDRESS, localKafkaUrl);
 
@@ -231,6 +235,7 @@ public void testGetChangelogConsumerThrowsException() {
   public void testGetStatefulChangelogConsumer()
       throws ExecutionException, InterruptedException, JsonProcessingException {
     Properties consumerProperties = new Properties();
+    consumerProperties.putAll(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs());
     String localKafkaUrl = "http://www.fooAddress.linkedin.com:16337";
     consumerProperties.put(KAFKA_BOOTSTRAP_SERVERS, localKafkaUrl);
     consumerProperties.put(CLUSTER_NAME, TEST_CLUSTER_NAME);
@@ -312,6 +317,7 @@ public void testGetStatefulChangelogConsumer()
   @Test
   public void testGetStatefulChangelogConsumerThrowsException() {
     Properties consumerProperties = new Properties();
+    consumerProperties.putAll(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs());
     String localKafkaUrl = "http://www.fooAddress.linkedin.com:16337";
     consumerProperties.put(KAFKA_BOOTSTRAP_SERVERS, localKafkaUrl);
     consumerProperties.put(CLUSTER_NAME, TEST_CLUSTER_NAME);
@@ -365,6 +371,7 @@ public void testCreatePubSubMessageDeserializer(
       boolean expectKmeWithSchemaReaderCall) {
     // Build properties
     Properties props = new Properties();
+    props.putAll(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs());
     if (kmeProp != null) {
       props.put(ConfigKeys.KME_SCHEMA_READER_FOR_SCHEMA_EVOLUTION_ENABLED, kmeProp);
     }

clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerDaVinciRecordTransformerImplTest.java

@@ -55,7 +55,6 @@
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
-import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
@@ -116,7 +115,7 @@ public void setUp() throws NoSuchFieldException, IllegalAccessException {
         .setStoreName(TEST_STORE_NAME)
         .setControllerD2ServiceName(D2_SERVICE_NAME)
         .setD2ServiceName(DEFAULT_CLUSTER_DISCOVERY_D2_SERVICE_NAME)
-        .setConsumerProperties(new Properties())
+        .setConsumerProperties(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs())
         .setLocalD2ZkHosts(TEST_ZOOKEEPER_ADDRESS)
         .setDatabaseSyncBytesInterval(TEST_DB_SYNC_BYTES_INTERVAL)
         .setD2Client(mock(D2Client.class))

clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerImplTest.java

@@ -208,6 +208,7 @@ public void testConfig() {
     assertTrue(config.getConsumerProperties().isEmpty());
     assertThrows(NullPointerException.class, () -> config.setConsumerProperties(null));
     Properties newProps = new Properties();
+    newProps.putAll(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs());
     newProps.setProperty("foo", "bar");
     config.setConsumerProperties(newProps);
     assertNotNull(config.getConsumerProperties());
@@ -1467,7 +1468,7 @@ private ChangelogClientConfig getChangelogClientConfig() {
         new ChangelogClientConfig<>().setD2ControllerClient(mockD2ControllerClient)
             .setSchemaReader(schemaReader)
             .setStoreName(storeName)
-            .setConsumerProperties(new Properties())
+            .setConsumerProperties(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs())
             .setViewName("");
     changelogClientConfig.getInnerClientConfig()
         .setMetricsRepository(getVeniceMetricsRepository(CHANGE_DATA_CAPTURE_CLIENT, CONSUMER_METRIC_ENTITIES, true));

clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VersionSpecificVeniceChangelogConsumerDaVinciRecordTransformerImplTest.java

@@ -47,7 +47,6 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
@@ -103,7 +102,7 @@ public void setUp() throws NoSuchFieldException, IllegalAccessException {
         .setStoreName(TEST_STORE_NAME)
         .setControllerD2ServiceName(D2_SERVICE_NAME)
         .setD2ServiceName(DEFAULT_CLUSTER_DISCOVERY_D2_SERVICE_NAME)
-        .setConsumerProperties(new Properties())
+        .setConsumerProperties(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs())
         .setLocalD2ZkHosts(TEST_ZOOKEEPER_ADDRESS)
         .setDatabaseSyncBytesInterval(TEST_DB_SYNC_BYTES_INTERVAL)
         .setD2Client(mock(D2Client.class))

clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTaskTest.java

@@ -3298,6 +3298,7 @@ public void testPartitionExceptionIsolation(AAConfig aaConfig) throws Exception
 
   private VeniceServerConfig buildVeniceServerConfig(Map<String, Object> extraProperties) {
     PropertyBuilder propertyBuilder = new PropertyBuilder();
+    propertyBuilder.put(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs());
     propertyBuilder.put(CLUSTER_NAME, "");
     propertyBuilder.put(ZOOKEEPER_ADDRESS, "");
     propertyBuilder.put(SERVER_PROMOTION_TO_LEADER_REPLICA_DELAY_SECONDS, 500L);

clients/da-vinci-client/src/test/java/com/linkedin/davinci/store/AbstractStorageEngineTest.java

@@ -17,6 +17,7 @@
 import com.linkedin.venice.meta.PersistenceType;
 import com.linkedin.venice.utils.PropertyBuilder;
 import com.linkedin.venice.utils.RandomGenUtils;
+import com.linkedin.venice.utils.TestUtils;
 import com.linkedin.venice.utils.Utils;
 import com.linkedin.venice.utils.VeniceProperties;
 import java.io.File;
@@ -45,6 +46,7 @@ public static VeniceProperties getServerProperties(PersistenceType persistenceTy
         .put(LISTENER_PORT, 7072)
         .put(ADMIN_PORT, 7073)
         .put(DATA_BASE_PATH, dataDirectory.getAbsolutePath())
+        .put(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs())
         .put(properties)
         .build();
   }

clients/da-vinci-client/src/test/resources/config/server.properties

@@ -1,4 +1,10 @@
 node.id=0
 listener.port=7072
 admin.port=7073
-kafka.threads.per.partition=1
+kafka.threads.per.partition=1
+# Pub-sub adapter factory classes (Apache Kafka). Required because the factory fails fast when the
+# class is unset and pubsub.adapter.factory.kafka.fallback.enabled is false (the default).
+pubsub.producer.adapter.factory.class=com.linkedin.venice.pubsub.adapter.kafka.producer.ApacheKafkaProducerAdapterFactory
+pubsub.consumer.adapter.factory.class=com.linkedin.venice.pubsub.adapter.kafka.consumer.ApacheKafkaConsumerAdapterFactory
+pubsub.admin.adapter.factory.class=com.linkedin.venice.pubsub.adapter.kafka.admin.ApacheKafkaAdminAdapterFactory
+pubsub.source.of.truth.admin.adapter.factory.class=com.linkedin.venice.pubsub.adapter.kafka.admin.ApacheKafkaAdminAdapterFactory

clients/venice-admin-tool/src/test/java/com/linkedin/venice/TestAdminTool.java

@@ -57,6 +57,7 @@
 import com.linkedin.venice.serialization.avro.AvroProtocolDefinition;
 import com.linkedin.venice.serializer.FastSerializerDeserializerFactory;
 import com.linkedin.venice.serializer.RecordSerializer;
+import com.linkedin.venice.utils.TestUtils;
 import com.linkedin.venice.utils.VeniceProperties;
 import com.linkedin.venice.views.MaterializedView;
 import java.io.IOException;
@@ -76,10 +77,40 @@
 import org.mockito.MockedStatic;
 import org.mockito.Mockito;
 import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
 
 
 public class TestAdminTool {
+  private static final String[] PUBSUB_ADAPTER_FACTORY_CONFIG_KEYS =
+      TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs().stringPropertyNames().toArray(new String[0]);
+  private static final Properties ORIGINAL_PUBSUB_ADAPTER_FACTORY_SYSTEM_PROPERTIES = new Properties();
+
+  @BeforeClass(alwaysRun = true)
+  public void setUpPubSubAdapterFactorySystemProperties() {
+    Properties factoryConfigs = TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs();
+    for (String key: PUBSUB_ADAPTER_FACTORY_CONFIG_KEYS) {
+      String originalValue = System.getProperty(key);
+      if (originalValue != null) {
+        ORIGINAL_PUBSUB_ADAPTER_FACTORY_SYSTEM_PROPERTIES.setProperty(key, originalValue);
+      }
+      System.setProperty(key, factoryConfigs.getProperty(key));
+    }
+  }
+
+  @AfterClass(alwaysRun = true)
+  public void restorePubSubAdapterFactorySystemProperties() {
+    for (String key: PUBSUB_ADAPTER_FACTORY_CONFIG_KEYS) {
+      if (ORIGINAL_PUBSUB_ADAPTER_FACTORY_SYSTEM_PROPERTIES.containsKey(key)) {
+        System.setProperty(key, ORIGINAL_PUBSUB_ADAPTER_FACTORY_SYSTEM_PROPERTIES.getProperty(key));
+      } else {
+        System.clearProperty(key);
+      }
+    }
+    ORIGINAL_PUBSUB_ADAPTER_FACTORY_SYSTEM_PROPERTIES.clear();
+  }
+
   @Test
   public void testPrintObject() {
     List<String> output = new ArrayList<>();