diff --git a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerImpl.java b/clients/da-vinci-client/src/main/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerImpl.java index 1d64062081c..56e07a446a4 100644 --- a/clients/da-vinci-client/src/main/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerImpl.java +++ b/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); diff --git a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/DaVinciBackendTest.java b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/DaVinciBackendTest.java index 2b581b70642..12ecb361c29 100644 --- a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/DaVinciBackendTest.java +++ b/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); diff --git a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/StoreBackendTest.java b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/StoreBackendTest.java index fc2273ca23b..2bcb871c425 100644 --- a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/StoreBackendTest.java +++ b/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()) diff --git a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/VersionBackendTest.java b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/VersionBackendTest.java index 55252da2c01..b4e6a1eba5f 100644 --- a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/VersionBackendTest.java +++ b/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); diff --git a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/client/AvroGenericDaVinciClientTest.java b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/client/AvroGenericDaVinciClientTest.java index 664d1ce6c1f..7d112ae4bac 100644 --- a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/client/AvroGenericDaVinciClientTest.java +++ b/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 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 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 dvcClient = diff --git a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/config/VeniceServerConfigTest.java b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/config/VeniceServerConfigTest.java index 9135195b45c..567415004e7 100644 --- a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/config/VeniceServerConfigTest.java +++ b/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"); diff --git a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/ChangelogClientConfigTest.java b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/ChangelogClientConfigTest.java index fb7dd603a35..6bc1e7cea8a 100644 --- a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/ChangelogClientConfigTest.java +++ b/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]; diff --git a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerClientFactoryTest.java b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerClientFactoryTest.java index 1ab53dbc84c..bda108b0791 100644 --- a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerClientFactoryTest.java +++ b/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); } diff --git a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerDaVinciRecordTransformerImplTest.java b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerDaVinciRecordTransformerImplTest.java index 41ac7dc16ac..01ceb499be2 100644 --- a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerDaVinciRecordTransformerImplTest.java +++ b/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)) diff --git a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerImplTest.java b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerImplTest.java index 4b1a8043736..6a6d876ce53 100644 --- a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerImplTest.java +++ b/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)); diff --git a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VersionSpecificVeniceChangelogConsumerDaVinciRecordTransformerImplTest.java b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VersionSpecificVeniceChangelogConsumerDaVinciRecordTransformerImplTest.java index c7e042c1ab4..edf78994945 100644 --- a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VersionSpecificVeniceChangelogConsumerDaVinciRecordTransformerImplTest.java +++ b/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)) diff --git a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTaskTest.java b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTaskTest.java index d4d60f3be06..3d72063ad7a 100644 --- a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTaskTest.java +++ b/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 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); diff --git a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/store/AbstractStorageEngineTest.java b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/store/AbstractStorageEngineTest.java index 7a22e97ec2b..01dcf7f7932 100644 --- a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/store/AbstractStorageEngineTest.java +++ b/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(); } diff --git a/clients/da-vinci-client/src/test/resources/config/server.properties b/clients/da-vinci-client/src/test/resources/config/server.properties index a2ed1f4a402..8ce5388f357 100644 --- a/clients/da-vinci-client/src/test/resources/config/server.properties +++ b/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 \ No newline at end of file +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 \ No newline at end of file diff --git a/clients/venice-admin-tool/src/test/java/com/linkedin/venice/TestAdminTool.java b/clients/venice-admin-tool/src/test/java/com/linkedin/venice/TestAdminTool.java index 4fd08c64308..0083f6904d7 100644 --- a/clients/venice-admin-tool/src/test/java/com/linkedin/venice/TestAdminTool.java +++ b/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 output = new ArrayList<>(); diff --git a/clients/venice-push-job/src/main/java/com/linkedin/venice/heartbeat/DefaultPushJobHeartbeatSenderFactory.java b/clients/venice-push-job/src/main/java/com/linkedin/venice/heartbeat/DefaultPushJobHeartbeatSenderFactory.java index ab2ec4d4f29..c79bd1aa432 100644 --- a/clients/venice-push-job/src/main/java/com/linkedin/venice/heartbeat/DefaultPushJobHeartbeatSenderFactory.java +++ b/clients/venice-push-job/src/main/java/com/linkedin/venice/heartbeat/DefaultPushJobHeartbeatSenderFactory.java @@ -1,6 +1,7 @@ package com.linkedin.venice.heartbeat; import static com.linkedin.venice.ConfigKeys.KAFKA_BOOTSTRAP_SERVERS; +import static com.linkedin.venice.ConfigKeys.PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS; import static com.linkedin.venice.status.BatchJobHeartbeatConfigs.HEARTBEAT_INITIAL_DELAY_CONFIG; import static com.linkedin.venice.status.BatchJobHeartbeatConfigs.HEARTBEAT_INTERVAL_CONFIG; import static com.linkedin.venice.status.BatchJobHeartbeatConfigs.HEARTBEAT_LAST_HEARTBEAT_IS_DELETE_CONFIG; @@ -61,7 +62,7 @@ public PushJobHeartbeatSender createHeartbeatSender( VeniceWriter veniceWriter = getVeniceWriter( heartbeatKafkaTopicName, partitionerConfig, - getVeniceWriterProperties(sslProperties, kafkaUrl), + getVeniceWriterProperties(sslProperties, kafkaUrl, properties), partitionNum); Schema heartbeatKeySchema = getHeartbeatKeySchema(controllerClient, retryAttempts, heartbeatStoreName); Map valueSchemasById = @@ -91,13 +92,22 @@ public PushJobHeartbeatSender createHeartbeatSender( return defaultPushJobHeartbeatSender; } - private Properties getVeniceWriterProperties(Optional sslProperties, String kafkaBootstrapUrl) { + private Properties getVeniceWriterProperties( + Optional sslProperties, + String kafkaBootstrapUrl, + VeniceProperties jobProperties) { Properties veniceWriterProperties = new Properties(); veniceWriterProperties.put(KAFKA_BOOTSTRAP_SERVERS, kafkaBootstrapUrl); if (sslProperties.isPresent()) { veniceWriterProperties.putAll(sslProperties.get()); } + // Forward the pub-sub producer adapter factory class so the heartbeat VeniceWriterFactory can resolve + // it; the factory fails fast otherwise (there is no implicit default). + if (jobProperties.containsKey(PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS)) { + veniceWriterProperties + .put(PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS, jobProperties.getString(PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS)); + } return veniceWriterProperties; } diff --git a/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/mapreduce/datawriter/reduce/TestVeniceReducer.java b/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/mapreduce/datawriter/reduce/TestVeniceReducer.java index 77145c00b45..16857d6a745 100644 --- a/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/mapreduce/datawriter/reduce/TestVeniceReducer.java +++ b/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/mapreduce/datawriter/reduce/TestVeniceReducer.java @@ -1,5 +1,7 @@ package com.linkedin.venice.hadoop.mapreduce.datawriter.reduce; +import static com.linkedin.venice.ConfigKeys.PASS_THROUGH_CONFIG_PREFIXES_LIST_KEY; +import static com.linkedin.venice.ConfigKeys.PUBSUB_CLIENT_CONFIG_PREFIX; import static com.linkedin.venice.ConfigKeys.PUSH_JOB_VIEW_CONFIGS; import static com.linkedin.venice.hadoop.mapreduce.counter.MRJobCounterHelper.TOTAL_KEY_SIZE_GROUP_COUNTER_NAME; import static com.linkedin.venice.hadoop.mapreduce.counter.MRJobCounterHelper.TOTAL_VALUE_SIZE_GROUP_COUNTER_NAME; @@ -52,6 +54,7 @@ import com.linkedin.venice.pubsub.api.PubSubProduceResult; import com.linkedin.venice.pubsub.api.PubSubProducerCallback; import com.linkedin.venice.serialization.avro.VeniceAvroKafkaSerializer; +import com.linkedin.venice.utils.TestUtils; import com.linkedin.venice.utils.VeniceProperties; import com.linkedin.venice.views.MaterializedView; import com.linkedin.venice.views.VeniceView; @@ -123,6 +126,8 @@ public void testReducerUpdateWithTooLargeValueAndChunkingDisabled() { private VeniceProperties getTestProps() { Properties props = new Properties(); + props.putAll(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs()); + props.put(PASS_THROUGH_CONFIG_PREFIXES_LIST_KEY, PUBSUB_CLIENT_CONFIG_PREFIX); props.put(MAP_REDUCE_JOB_ID_PROP, "job_200707121733_0003"); props.put(VALUE_SCHEMA_ID_PROP, 1); props.put(VENICE_PUSH_DESTINATION_PUBSUB_BROKER, "localhost:8090"); /* Destination Kafka cluster */ diff --git a/clients/venice-push-job/src/test/java/com/linkedin/venice/heartbeat/TestPushJobHeartbeatSender.java b/clients/venice-push-job/src/test/java/com/linkedin/venice/heartbeat/TestPushJobHeartbeatSender.java index 6a25bb9a758..646aacf7e61 100644 --- a/clients/venice-push-job/src/test/java/com/linkedin/venice/heartbeat/TestPushJobHeartbeatSender.java +++ b/clients/venice-push-job/src/test/java/com/linkedin/venice/heartbeat/TestPushJobHeartbeatSender.java @@ -15,6 +15,7 @@ import com.linkedin.venice.serialization.avro.AvroProtocolDefinition; import com.linkedin.venice.status.protocol.BatchJobHeartbeatKey; import com.linkedin.venice.status.protocol.BatchJobHeartbeatValue; +import com.linkedin.venice.utils.TestUtils; import com.linkedin.venice.utils.Utils; import com.linkedin.venice.utils.VeniceProperties; import java.util.Optional; @@ -28,7 +29,9 @@ public class TestPushJobHeartbeatSender { public void testHeartbeatSenderCreation() { String kafkaUrl = "localhost:1234"; String heartbeatStoreName = AvroProtocolDefinition.BATCH_JOB_HEARTBEAT.getSystemStoreName(); - VeniceProperties properties = VeniceProperties.empty(); + VeniceProperties properties = new VeniceProperties(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs()); + // No SSL properties: the heartbeat writer must still resolve the pub-sub producer factory class from + // the job properties (see DefaultPushJobHeartbeatSenderFactory), which is the sslEnabled=false path. Optional sslProperties = Optional.empty(); DefaultPushJobHeartbeatSenderFactory pushJobHeartbeatSenderFactory = new DefaultPushJobHeartbeatSenderFactory(); diff --git a/docker/venice-controller/multi-dc-configs/dc-0.venice.controller.properties b/docker/venice-controller/multi-dc-configs/dc-0.venice.controller.properties index 9ff7fe9fa4e..e2ce3b607dc 100644 --- a/docker/venice-controller/multi-dc-configs/dc-0.venice.controller.properties +++ b/docker/venice-controller/multi-dc-configs/dc-0.venice.controller.properties @@ -55,3 +55,10 @@ active.active.real.time.source.fabric.list=dc-0,dc-1 controller.enable.batch.push.from.admin.in.child=false default.partition.size=100 topic.cleanup.delay.factor=2 + +# 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 diff --git a/docker/venice-controller/multi-dc-configs/dc-parent.venice.controller.properties b/docker/venice-controller/multi-dc-configs/dc-parent.venice.controller.properties index 1db1dc23f4f..828965b709f 100644 --- a/docker/venice-controller/multi-dc-configs/dc-parent.venice.controller.properties +++ b/docker/venice-controller/multi-dc-configs/dc-parent.venice.controller.properties @@ -56,3 +56,10 @@ kafka.replication.factor=1 native.replication.source.fabric.as.default.for.batch.only.stores=dc-0 default.partition.size=100 topic.cleanup.delay.factor=2 + +# 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 diff --git a/docker/venice-controller/single-dc-configs/controller.properties b/docker/venice-controller/single-dc-configs/controller.properties index 0e2422d8fb9..9aff1f261fd 100644 --- a/docker/venice-controller/single-dc-configs/controller.properties +++ b/docker/venice-controller/single-dc-configs/controller.properties @@ -27,3 +27,10 @@ enable.offline.push.ssl.whitelist=false kafka.linger.ms=0 default.partition.count=1 controller.zk.shared.metadata.system.schema.store.auto.creation.enabled=true + +# 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 diff --git a/docker/venice-server/multi-dc-configs/dc-0/server.properties b/docker/venice-server/multi-dc-configs/dc-0/server.properties index e96f60cc807..a9fec9041b2 100644 --- a/docker/venice-server/multi-dc-configs/dc-0/server.properties +++ b/docker/venice-server/multi-dc-configs/dc-0/server.properties @@ -16,3 +16,9 @@ persistence.type=ROCKS_DB rocksdb.block.cache.size.in.bytes=2147483648 rocksdb.sst.file.manager.delete.rate.bytes.per.second=524288000 rocksdb.sst.file.manager.max.trash.db.ratio=0.25 + +# 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 diff --git a/docker/venice-server/multi-dc-configs/dc-1/server.properties b/docker/venice-server/multi-dc-configs/dc-1/server.properties index 40e4190f677..472d678eb45 100644 --- a/docker/venice-server/multi-dc-configs/dc-1/server.properties +++ b/docker/venice-server/multi-dc-configs/dc-1/server.properties @@ -16,3 +16,9 @@ persistence.type=ROCKS_DB rocksdb.block.cache.size.in.bytes=2147483648 rocksdb.sst.file.manager.delete.rate.bytes.per.second=524288000 rocksdb.sst.file.manager.max.trash.db.ratio=0.25 + +# 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 diff --git a/docker/venice-server/single-dc-configs/server.properties b/docker/venice-server/single-dc-configs/server.properties index 4d00fa10887..9122dd07a05 100644 --- a/docker/venice-server/single-dc-configs/server.properties +++ b/docker/venice-server/single-dc-configs/server.properties @@ -19,3 +19,9 @@ persistence.type=ROCKS_DB rocksdb.block.cache.size.in.bytes=2147483648 rocksdb.sst.file.manager.delete.rate.bytes.per.second=524288000 rocksdb.sst.file.manager.max.trash.db.ratio=0.25 + +# 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 diff --git a/integrations/venice-duckdb/src/integrationTest/java/com/linkedin/venice/endToEnd/DuckDBDaVinciRecordTransformerIntegrationTest.java b/integrations/venice-duckdb/src/integrationTest/java/com/linkedin/venice/endToEnd/DuckDBDaVinciRecordTransformerIntegrationTest.java index ff24637a106..141a5151a16 100644 --- a/integrations/venice-duckdb/src/integrationTest/java/com/linkedin/venice/endToEnd/DuckDBDaVinciRecordTransformerIntegrationTest.java +++ b/integrations/venice-duckdb/src/integrationTest/java/com/linkedin/venice/endToEnd/DuckDBDaVinciRecordTransformerIntegrationTest.java @@ -172,7 +172,7 @@ public void testRecordTransformer() throws Exception { ClientConfig.defaultGenericClientConfig(storeName) .setD2Client(d2Client) .setD2ServiceName(VeniceRouterWrapper.CLUSTER_DISCOVERY_D2_SERVICE_NAME), - VeniceProperties.empty(), + new VeniceProperties(ServiceFactory.getPubSubClientConfigs()), null)) { producer.asyncDelete(getKey(1)).get(); } diff --git a/integrations/venice-pulsar/src/main/java/com/linkedin/venice/pulsar/sink/VenicePulsarSink.java b/integrations/venice-pulsar/src/main/java/com/linkedin/venice/pulsar/sink/VenicePulsarSink.java index fa5fc1ed81f..51e730314ea 100644 --- a/integrations/venice-pulsar/src/main/java/com/linkedin/venice/pulsar/sink/VenicePulsarSink.java +++ b/integrations/venice-pulsar/src/main/java/com/linkedin/venice/pulsar/sink/VenicePulsarSink.java @@ -9,6 +9,7 @@ import static com.linkedin.venice.samza.VeniceSystemFactory.VENICE_ROUTER_URL; import static com.linkedin.venice.samza.VeniceSystemFactory.VENICE_STORE; +import com.linkedin.venice.ConfigKeys; import com.linkedin.venice.meta.Version; import com.linkedin.venice.samza.VeniceSystemFactory; import com.linkedin.venice.samza.VeniceSystemProducer; @@ -242,6 +243,11 @@ public static Map getConfig(VenicePulsarSinkConfig veniceCfg, St config.put(VENICE_ROUTER_URL, veniceCfg.getVeniceRouterUrl()); config.put(DEPLOYMENT_ID, Utils.getUniqueString("venice-push-id-pulsar-sink")); config.put(SSL_ENABLED, "false"); + // The sink's Venice producer writes to the (Kafka-backed) real-time topic; set the pub-sub producer + // adapter factory explicitly so it does not fail fast when the factory class is unconfigured. + config.put( + ConfigKeys.PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS, + "com.linkedin.venice.pubsub.adapter.kafka.producer.ApacheKafkaProducerAdapterFactory"); if (veniceCfg.getKafkaSaslConfig() != null && !veniceCfg.getKafkaSaslConfig().isEmpty()) { config.put("kafka.sasl.jaas.config", veniceCfg.getKafkaSaslConfig()); } diff --git a/integrations/venice-samza/src/test/java/com/linkedin/venice/samza/VeniceSystemProducerTest.java b/integrations/venice-samza/src/test/java/com/linkedin/venice/samza/VeniceSystemProducerTest.java index 20c0971d8c9..058c534f7d5 100644 --- a/integrations/venice-samza/src/test/java/com/linkedin/venice/samza/VeniceSystemProducerTest.java +++ b/integrations/venice-samza/src/test/java/com/linkedin/venice/samza/VeniceSystemProducerTest.java @@ -2,6 +2,7 @@ import static com.linkedin.venice.CommonConfigKeys.SSL_ENABLED; import static com.linkedin.venice.ConfigKeys.KAFKA_BOOTSTRAP_SERVERS; +import static com.linkedin.venice.ConfigKeys.PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS; import static com.linkedin.venice.ConfigKeys.VALIDATE_VENICE_INTERNAL_SCHEMA_VERSION; import static com.linkedin.venice.ConfigKeys.VENICE_PARTITIONERS; import static com.linkedin.venice.VeniceConstants.SYSTEM_PROPERTY_FOR_APP_RUNNING_REGION; @@ -35,6 +36,7 @@ import com.linkedin.venice.meta.StoreInfo; import com.linkedin.venice.meta.Version; import com.linkedin.venice.meta.VersionImpl; +import com.linkedin.venice.pubsub.adapter.kafka.producer.ApacheKafkaProducerAdapterFactory; import com.linkedin.venice.pubsub.api.PubSubProducerAdapter; import com.linkedin.venice.pushmonitor.ExecutionStatus; import com.linkedin.venice.pushmonitor.RouterBasedPushMonitor; @@ -48,12 +50,14 @@ import com.linkedin.venice.writer.update.UpdateBuilderImpl; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.Properties; import org.apache.avro.Schema; import org.apache.avro.generic.GenericRecord; import org.apache.samza.SamzaException; import org.apache.samza.config.Config; +import org.apache.samza.config.MapConfig; import org.apache.samza.system.OutgoingMessageEnvelope; import org.apache.samza.system.SystemProducer; import org.apache.samza.system.SystemStream; @@ -72,6 +76,11 @@ public void testPartialUpdateConversion() { .setSamzaJobId("push-job-id-1") .setRunningFabric("dc-0") .setFactory(mock(VeniceSystemFactory.class)) + .setSamzaConfig( + new MapConfig( + Collections.singletonMap( + PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS, + ApacheKafkaProducerAdapterFactory.class.getName()))) .setVeniceChildD2ZkHost("zookeeper.com:2181") .setPrimaryControllerColoD2ZKHost("zookeeper.com:2181") .setPrimaryControllerD2ServiceName("ChildController") @@ -142,6 +151,11 @@ public void testGetVeniceWriter(Version.PushType pushType) { .setSamzaJobId("push-job-id-1") .setRunningFabric("dc-0") .setFactory(mock(VeniceSystemFactory.class)) + .setSamzaConfig( + new MapConfig( + Collections.singletonMap( + PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS, + ApacheKafkaProducerAdapterFactory.class.getName()))) .setVeniceChildD2ZkHost("zookeeper.com:2181") .setPrimaryControllerColoD2ZKHost("zookeeper.com:2181") .setPrimaryControllerD2ServiceName("ChildController") @@ -171,6 +185,9 @@ public void testGetVeniceWriter(Version.PushType pushType) { assertNotNull(resultantVeniceWriter); assertEquals(resultantVeniceWriter, veniceWriterMock); assertEquals(capturedProperties.getProperty(KAFKA_BOOTSTRAP_SERVERS), "venice-kafka.db:2023"); + assertEquals( + capturedProperties.getProperty(PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS), + ApacheKafkaProducerAdapterFactory.class.getName()); assertEquals(capturedVwo.getTopicName(), "test_store_v1"); if (pushType != Version.PushType.BATCH && pushType != Version.PushType.STREAM_REPROCESSING) { // invoke create venice write without partition count diff --git a/internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java b/internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java index d8816497ff5..b5330384316 100644 --- a/internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java +++ b/internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java @@ -116,6 +116,19 @@ private ConfigKeys() { public static final String PUBSUB_SOURCE_OF_TRUTH_ADMIN_ADAPTER_FACTORY_CLASS = PUBSUB_CLIENT_CONFIG_PREFIX + "source.of.truth.admin.adapter.factory.class"; + /** + * Configuration key that controls whether the PubSub producer/consumer/admin adapter factories + * silently fall back to the Apache Kafka implementation when their factory-class config keys are + * not explicitly provided. + *

+ * When {@code false} (the default), the {@code PubSubClientsFactory} fails fast by throwing an + * exception if the corresponding factory-class config is missing. This surfaces misconfiguration + * early instead of masking it behind an implicit Kafka default. Set this to {@code true} to + * restore the legacy behavior of defaulting to the Apache Kafka adapter factories. + */ + public static final String PUBSUB_ADAPTER_FACTORY_KAFKA_FALLBACK_ENABLED = + PUBSUB_CLIENT_CONFIG_PREFIX + "adapter.factory.kafka.fallback.enabled"; + /** * Configuration key for specifying the address of the PubSub broker (e.g., Kafka, Pulsar). *

diff --git a/internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubClientsFactory.java b/internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubClientsFactory.java index cf376c2d178..e68221cf25a 100644 --- a/internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubClientsFactory.java +++ b/internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubClientsFactory.java @@ -1,5 +1,6 @@ package com.linkedin.venice.pubsub; +import static com.linkedin.venice.ConfigKeys.PUBSUB_ADAPTER_FACTORY_KAFKA_FALLBACK_ENABLED; import static com.linkedin.venice.ConfigKeys.PUBSUB_ADMIN_ADAPTER_FACTORY_CLASS; import static com.linkedin.venice.ConfigKeys.PUBSUB_CONSUMER_ADAPTER_FACTORY_CLASS; import static com.linkedin.venice.ConfigKeys.PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS; @@ -29,6 +30,14 @@ public class PubSubClientsFactory { private static final Logger LOGGER = LogManager.getLogger(PubSubClientsFactory.class); + /** + * By default the adapter factories do NOT fall back to Apache Kafka when their factory-class config + * is missing; callers must configure the factory classes explicitly so that misconfiguration fails + * fast. Set {@link com.linkedin.venice.ConfigKeys#PUBSUB_ADAPTER_FACTORY_KAFKA_FALLBACK_ENABLED} to + * {@code true} to restore the legacy implicit-Kafka behavior. + */ + public static final boolean DEFAULT_KAFKA_FALLBACK_ENABLED = false; + private enum FactoryType { PRODUCER, CONSUMER, ADMIN } @@ -107,13 +116,28 @@ private static T createFactory( String alternateConfigKey, String defaultClassName, FactoryType factoryType) { - String className; - if (properties.containsKey(preferredConfigKey) || properties.containsKey(alternateConfigKey)) { - className = properties.getStringWithAlternative(preferredConfigKey, alternateConfigKey); - LOGGER.debug("Creating pub-sub {} adapter factory instance for class: {}", factoryType, className); - } else { + String className = properties.getStringWithAlternative(preferredConfigKey, alternateConfigKey, null); + if (className == null || className.trim().isEmpty()) { + // A missing or blank value is treated as "not configured": either fall back to the Apache Kafka + // adapter factory (when explicitly enabled) or fail fast with a message naming the missing key. + boolean kafkaFallbackEnabled = + properties.getBoolean(PUBSUB_ADAPTER_FACTORY_KAFKA_FALLBACK_ENABLED, DEFAULT_KAFKA_FALLBACK_ENABLED); + if (!kafkaFallbackEnabled) { + throw new VeniceException( + String.format( + "PubSub %s adapter factory class is not configured. Set '%s' (or the legacy '%s') to the " + + "fully-qualified factory class name. Implicit fallback to the Apache Kafka adapter factory " + + "('%s') is disabled; set '%s=true' to re-enable it.", + factoryType, + preferredConfigKey, + alternateConfigKey, + defaultClassName, + PUBSUB_ADAPTER_FACTORY_KAFKA_FALLBACK_ENABLED)); + } className = defaultClassName; LOGGER.debug("Creating pub-sub {} adapter factory instance with default class: {}", factoryType, className); + } else { + LOGGER.debug("Creating pub-sub {} adapter factory instance for class: {}", factoryType, className); } return createInstance(className); diff --git a/internal/venice-common/src/test/java/com/linkedin/venice/pubsub/PubSubClientsFactoryTest.java b/internal/venice-common/src/test/java/com/linkedin/venice/pubsub/PubSubClientsFactoryTest.java index 02aff808438..8bf87d00389 100644 --- a/internal/venice-common/src/test/java/com/linkedin/venice/pubsub/PubSubClientsFactoryTest.java +++ b/internal/venice-common/src/test/java/com/linkedin/venice/pubsub/PubSubClientsFactoryTest.java @@ -1,13 +1,16 @@ package com.linkedin.venice.pubsub; +import static com.linkedin.venice.ConfigKeys.PUBSUB_ADAPTER_FACTORY_KAFKA_FALLBACK_ENABLED; import static com.linkedin.venice.ConfigKeys.PUBSUB_ADMIN_ADAPTER_FACTORY_CLASS; import static com.linkedin.venice.ConfigKeys.PUBSUB_CONSUMER_ADAPTER_FACTORY_CLASS; import static com.linkedin.venice.ConfigKeys.PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS; +import static com.linkedin.venice.ConfigKeys.PUBSUB_SOURCE_OF_TRUTH_ADMIN_ADAPTER_FACTORY_CLASS; import static com.linkedin.venice.ConfigKeys.PUB_SUB_ADMIN_ADAPTER_FACTORY_CLASS; import static com.linkedin.venice.ConfigKeys.PUB_SUB_CONSUMER_ADAPTER_FACTORY_CLASS; import static com.linkedin.venice.ConfigKeys.PUB_SUB_PRODUCER_ADAPTER_FACTORY_CLASS; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertTrue; import static org.testng.Assert.expectThrows; import com.linkedin.venice.exceptions.VeniceException; @@ -26,13 +29,6 @@ public class PubSubClientsFactoryTest { @Test public void testCreateInstanceSuccess() { - // default: no config provided - verifyFactoryClasses( - new Properties(), - ApacheKafkaProducerAdapterFactory.class, - ApacheKafkaConsumerAdapterFactory.class, - ApacheKafkaAdminAdapterFactory.class); - // with legacy config names Properties legacyProps = new Properties(); legacyProps.put(PUB_SUB_PRODUCER_ADAPTER_FACTORY_CLASS, TestPubSubProducerAdapterFactory.class.getName()); @@ -56,6 +52,75 @@ public void testCreateInstanceSuccess() { TestPubSubAdminAdapterFactory.class); } + /** + * By default (no factory-class config and no explicit fallback flag) the factory should fail fast + * instead of silently defaulting to the Apache Kafka adapter factories. + */ + @Test + public void testFailFastWhenFactoryClassMissingAndFallbackDisabled() { + VeniceProperties emptyProps = new VeniceProperties(new Properties()); + + assertFailFast(() -> PubSubClientsFactory.createProducerFactory(emptyProps), PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS); + assertFailFast(() -> PubSubClientsFactory.createConsumerFactory(emptyProps), PUBSUB_CONSUMER_ADAPTER_FACTORY_CLASS); + assertFailFast(() -> PubSubClientsFactory.createAdminFactory(emptyProps), PUBSUB_ADMIN_ADAPTER_FACTORY_CLASS); + assertFailFast( + () -> PubSubClientsFactory.createSourceOfTruthAdminFactory(emptyProps), + PUBSUB_SOURCE_OF_TRUTH_ADMIN_ADAPTER_FACTORY_CLASS); + // The instance constructor eagerly builds all three factories, so it should fail fast as well. + expectThrows(VeniceException.class, () -> new PubSubClientsFactory(emptyProps)); + + // Explicitly disabling the fallback behaves the same as the default. + Properties fallbackDisabled = new Properties(); + fallbackDisabled.put(PUBSUB_ADAPTER_FACTORY_KAFKA_FALLBACK_ENABLED, "false"); + expectThrows(VeniceException.class, () -> new PubSubClientsFactory(new VeniceProperties(fallbackDisabled))); + } + + /** + * A present-but-blank factory-class config is treated the same as a missing one: it fails fast with a + * message naming the key (instead of attempting to instantiate an empty class name), and resolves to + * the Apache Kafka default when the fallback is explicitly enabled. + */ + @Test + public void testBlankFactoryClassTreatedAsMissing() { + Properties blankProps = new Properties(); + blankProps.put(PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS, " "); + assertFailFast( + () -> PubSubClientsFactory.createProducerFactory(new VeniceProperties(blankProps)), + PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS); + + Properties blankWithFallback = new Properties(); + blankWithFallback.put(PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS, ""); + blankWithFallback.put(PUBSUB_ADAPTER_FACTORY_KAFKA_FALLBACK_ENABLED, "true"); + PubSubProducerAdapterFactory producerFactory = + PubSubClientsFactory.createProducerFactory(new VeniceProperties(blankWithFallback)); + assertEquals(producerFactory.getClass().getName(), ApacheKafkaProducerAdapterFactory.class.getName()); + } + + /** + * When the Kafka fallback is explicitly enabled, missing factory-class configs should resolve to the + * Apache Kafka adapter factories (the legacy behavior). + */ + @Test + public void testKafkaFallbackWhenExplicitlyEnabled() { + Properties fallbackEnabled = new Properties(); + fallbackEnabled.put(PUBSUB_ADAPTER_FACTORY_KAFKA_FALLBACK_ENABLED, "true"); + verifyFactoryClasses( + fallbackEnabled, + ApacheKafkaProducerAdapterFactory.class, + ApacheKafkaConsumerAdapterFactory.class, + ApacheKafkaAdminAdapterFactory.class); + } + + private static void assertFailFast(org.testng.Assert.ThrowingRunnable runnable, String expectedConfigKeyInMessage) { + VeniceException e = expectThrows(VeniceException.class, runnable); + assertTrue( + e.getMessage().contains(expectedConfigKeyInMessage), + "Expected fail-fast message to reference '" + expectedConfigKeyInMessage + "' but was: " + e.getMessage()); + assertTrue( + e.getMessage().contains(PUBSUB_ADAPTER_FACTORY_KAFKA_FALLBACK_ENABLED), + "Expected fail-fast message to reference the fallback config key but was: " + e.getMessage()); + } + private void verifyFactoryClasses( Properties props, Class expectedProducer, diff --git a/internal/venice-common/src/test/java/com/linkedin/venice/writer/VeniceWriterFactoryTest.java b/internal/venice-common/src/test/java/com/linkedin/venice/writer/VeniceWriterFactoryTest.java index aacb7c249be..2425ef0d00d 100644 --- a/internal/venice-common/src/test/java/com/linkedin/venice/writer/VeniceWriterFactoryTest.java +++ b/internal/venice-common/src/test/java/com/linkedin/venice/writer/VeniceWriterFactoryTest.java @@ -110,6 +110,7 @@ public void testVeniceWriterFactoryWithProducerCompressionDisabled() { public void testVeniceWriterFactoryCreatesProducerAdapterFactory() { Properties properties = new Properties(); properties.put(ConfigKeys.PUBSUB_BROKER_ADDRESS, "kafka:9898"); + properties.put(ConfigKeys.PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS, ApacheKafkaProducerAdapterFactory.class.getName()); VeniceWriterFactory veniceWriterFactory = new VeniceWriterFactory(properties, null, null, null); assertNotNull(veniceWriterFactory.getProducerAdapterFactory()); diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/davinci/DaVinciUserApp.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/davinci/DaVinciUserApp.java index 08fc76bc2fc..f76a552e6e8 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/davinci/DaVinciUserApp.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/davinci/DaVinciUserApp.java @@ -30,6 +30,7 @@ import com.linkedin.venice.D2.D2ClientUtils; import com.linkedin.venice.endToEnd.TestStringRecordTransformer; import com.linkedin.venice.integration.utils.DaVinciTestContext; +import com.linkedin.venice.integration.utils.ServiceFactory; import io.tehuti.metrics.MetricsRepository; import java.io.FileInputStream; import java.io.IOException; @@ -100,6 +101,9 @@ public static void main(String[] args) throws Exception { D2ClientUtils.startClient(d2Client); Map extraBackendConfig = new HashMap<>(); + // This forked DaVinci process must configure the pub-sub adapter factory classes explicitly; the + // factory fails fast otherwise (there is no implicit default). + ServiceFactory.getPubSubClientConfigs().forEach((key, value) -> extraBackendConfig.put(key.toString(), value)); extraBackendConfig.put(DATA_BASE_PATH, baseDataPath); extraBackendConfig.put(PUSH_STATUS_STORE_ENABLED, true); diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/consumer/ChangelogConsumerDaVinciRecordTransformerUserApp.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/consumer/ChangelogConsumerDaVinciRecordTransformerUserApp.java index 60f942eb963..8c05a600408 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/consumer/ChangelogConsumerDaVinciRecordTransformerUserApp.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/consumer/ChangelogConsumerDaVinciRecordTransformerUserApp.java @@ -43,6 +43,7 @@ import com.linkedin.venice.D2.D2ClientUtils; import com.linkedin.venice.endToEnd.TestChangelogKey; import com.linkedin.venice.endToEnd.TestChangelogValue; +import com.linkedin.venice.integration.utils.ServiceFactory; import com.linkedin.venice.integration.utils.VeniceRouterWrapper; import com.linkedin.venice.pubsub.api.PubSubMessage; import com.linkedin.venice.utils.SslUtils; @@ -91,6 +92,9 @@ public static void main(String[] args) throws InterruptedException, ExecutionExc getVeniceMetricsRepository(CHANGE_DATA_CAPTURE_CLIENT, CONSUMER_METRIC_ENTITIES, true); Properties consumerProperties = new Properties(); + // This forked process must configure the pub-sub adapter factory classes explicitly; the factory + // fails fast otherwise (there is no implicit default). + consumerProperties.putAll(ServiceFactory.getPubSubClientConfigs()); consumerProperties.put(KAFKA_BOOTSTRAP_SERVERS, kafkaUrl); consumerProperties.put(CLUSTER_NAME, clusterName); consumerProperties.put(ZOOKEEPER_ADDRESS, zkUrl); diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/consumer/ChangelogConsumerTestUtils.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/consumer/ChangelogConsumerTestUtils.java index 328f0d3a77d..6323c8b9879 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/consumer/ChangelogConsumerTestUtils.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/consumer/ChangelogConsumerTestUtils.java @@ -17,6 +17,7 @@ import com.linkedin.venice.controllerapi.ControllerClient; import com.linkedin.venice.controllerapi.UpdateStoreQueryParams; import com.linkedin.venice.integration.utils.PubSubBrokerWrapper; +import com.linkedin.venice.integration.utils.ServiceFactory; import com.linkedin.venice.integration.utils.VeniceClusterWrapper; import com.linkedin.venice.integration.utils.VeniceRouterWrapper; import com.linkedin.venice.integration.utils.VeniceTwoLayerMultiRegionMultiClusterWrapper; @@ -90,6 +91,7 @@ private static Properties buildConsumerProperties( String zkAddress) { Properties consumerProperties = new Properties(); consumerProperties.putAll(pubSubClientProperties); + consumerProperties.putAll(ServiceFactory.getPubSubClientConfigs()); consumerProperties.put(KAFKA_BOOTSTRAP_SERVERS, kafkaBootstrapServers); consumerProperties.put(CLUSTER_NAME, clusterName); consumerProperties.put(ZOOKEEPER_ADDRESS, zkAddress); diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/AbstractTestVeniceHelixAdmin.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/AbstractTestVeniceHelixAdmin.java index d5f0ddd5442..836ac237922 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/AbstractTestVeniceHelixAdmin.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/AbstractTestVeniceHelixAdmin.java @@ -330,6 +330,7 @@ void stopParticipant(String nodeId) { Properties getControllerProperties(String clusterName) throws IOException { Properties properties = TestUtils.getPropertiesForControllerConfig(); + properties.putAll(ServiceFactory.getPubSubClientConfigs()); properties.put(DEFAULT_OFFLINE_PUSH_STRATEGY, OfflinePushStrategy.WAIT_ALL_REPLICAS.name()); properties.put(DELAY_TO_REBALANCE_MS, 0); properties.put(KAFKA_REPLICATION_FACTOR, 1); diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/AdminToolE2ETest.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/AdminToolE2ETest.java index a79413fd1ba..ee343714c29 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/AdminToolE2ETest.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/AdminToolE2ETest.java @@ -53,9 +53,11 @@ public class AdminToolE2ETest { private List childDatacenters; private VeniceTwoLayerMultiRegionMultiClusterWrapper multiRegionMultiClusterWrapper; + private Properties originalPubSubAdapterFactorySystemProperties; @BeforeClass public void setUp() { + originalPubSubAdapterFactorySystemProperties = ServiceFactory.setPubSubClientConfigsAsSystemProperties(); // Disable auto materialization here as we need to test the back-fill command. Properties parentControllerProperties = new Properties(); parentControllerProperties.setProperty(CONTROLLER_AUTO_MATERIALIZE_META_SYSTEM_STORE, "false"); @@ -80,6 +82,13 @@ public void setUp() { @AfterClass(alwaysRun = true) public void cleanUp() { multiRegionMultiClusterWrapper.close(); + restorePubSubAdapterFactorySystemProperties(); + } + + private void restorePubSubAdapterFactorySystemProperties() { + if (originalPubSubAdapterFactorySystemProperties != null) { + ServiceFactory.restorePubSubClientConfigsSystemProperties(originalPubSubAdapterFactorySystemProperties); + } } @Test(timeOut = TEST_TIMEOUT) diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/TestAdminToolClusterConfig.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/TestAdminToolClusterConfig.java index 74a7fec3463..45f69de5ef6 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/TestAdminToolClusterConfig.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/TestAdminToolClusterConfig.java @@ -37,9 +37,11 @@ public class TestAdminToolClusterConfig { String clusterName; VeniceClusterWrapper venice; + private Properties originalPubSubAdapterFactorySystemProperties; @BeforeClass public void setUp() { + originalPubSubAdapterFactorySystemProperties = ServiceFactory.setPubSubClientConfigsAsSystemProperties(); Properties properties = new Properties(); String regionName = "dc-0"; properties.setProperty(LOCAL_REGION_NAME, regionName); @@ -67,9 +69,16 @@ public void setUp() { clusterName = venice.getClusterName(); } - @AfterClass + @AfterClass(alwaysRun = true) public void cleanUp() { venice.close(); + restorePubSubAdapterFactorySystemProperties(); + } + + private void restorePubSubAdapterFactorySystemProperties() { + if (originalPubSubAdapterFactorySystemProperties != null) { + ServiceFactory.restorePubSubClientConfigsSystemProperties(originalPubSubAdapterFactorySystemProperties); + } } @Test(timeOut = TEST_TIMEOUT) diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/TestAdminToolDataOperations.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/TestAdminToolDataOperations.java index 842eb34db9b..c3c472240ef 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/TestAdminToolDataOperations.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/TestAdminToolDataOperations.java @@ -48,9 +48,11 @@ public class TestAdminToolDataOperations { String clusterName; VeniceClusterWrapper venice; + private Properties originalPubSubAdapterFactorySystemProperties; @BeforeClass public void setUp() { + originalPubSubAdapterFactorySystemProperties = ServiceFactory.setPubSubClientConfigsAsSystemProperties(); Properties properties = new Properties(); String regionName = "dc-0"; properties.setProperty(LOCAL_REGION_NAME, regionName); @@ -78,9 +80,16 @@ public void setUp() { clusterName = venice.getClusterName(); } - @AfterClass + @AfterClass(alwaysRun = true) public void cleanUp() { venice.close(); + restorePubSubAdapterFactorySystemProperties(); + } + + private void restorePubSubAdapterFactorySystemProperties() { + if (originalPubSubAdapterFactorySystemProperties != null) { + ServiceFactory.restorePubSubClientConfigsSystemProperties(originalPubSubAdapterFactorySystemProperties); + } } @Test(timeOut = TEST_TIMEOUT * 4) diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/TestAdminToolEndToEnd.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/TestAdminToolEndToEnd.java index 84bdd5c4cbc..cf01aad93e3 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/TestAdminToolEndToEnd.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/TestAdminToolEndToEnd.java @@ -42,9 +42,11 @@ public class TestAdminToolEndToEnd { String clusterName; VeniceClusterWrapper venice; + private Properties originalPubSubAdapterFactorySystemProperties; @BeforeClass public void setUp() { + originalPubSubAdapterFactorySystemProperties = ServiceFactory.setPubSubClientConfigsAsSystemProperties(); Properties properties = new Properties(); String regionName = "dc-0"; properties.setProperty(LOCAL_REGION_NAME, regionName); @@ -72,9 +74,16 @@ public void setUp() { clusterName = venice.getClusterName(); } - @AfterClass + @AfterClass(alwaysRun = true) public void cleanUp() { venice.close(); + restorePubSubAdapterFactorySystemProperties(); + } + + private void restorePubSubAdapterFactorySystemProperties() { + if (originalPubSubAdapterFactorySystemProperties != null) { + ServiceFactory.restorePubSubClientConfigsSystemProperties(originalPubSubAdapterFactorySystemProperties); + } } @Test(timeOut = TEST_TIMEOUT) diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/TestFabricBuildout.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/TestFabricBuildout.java index e36468fcc7a..302fe772cba 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/TestFabricBuildout.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/TestFabricBuildout.java @@ -49,9 +49,11 @@ public class TestFabricBuildout { private List childDatacenters; private List parentControllers; private VeniceTwoLayerMultiRegionMultiClusterWrapper multiRegionMultiClusterWrapper; + private Properties originalPubSubAdapterFactorySystemProperties; @BeforeClass public void setUp() { + originalPubSubAdapterFactorySystemProperties = ServiceFactory.setPubSubClientConfigsAsSystemProperties(); Properties childControllerProperties = new Properties(); childControllerProperties.setProperty(ALLOW_CLUSTER_WIPE, "true"); Properties serverProperties = new Properties(); @@ -79,6 +81,13 @@ public void setUp() { @AfterClass(alwaysRun = true) public void cleanUp() { multiRegionMultiClusterWrapper.close(); + restorePubSubAdapterFactorySystemProperties(); + } + + private void restorePubSubAdapterFactorySystemProperties() { + if (originalPubSubAdapterFactorySystemProperties != null) { + ServiceFactory.restorePubSubClientConfigsSystemProperties(originalPubSubAdapterFactorySystemProperties); + } } @Test(timeOut = TEST_TIMEOUT) diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/AbstractMultiRegionTest.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/AbstractMultiRegionTest.java index 32ce6c9ba2b..391d79ef0d2 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/AbstractMultiRegionTest.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/AbstractMultiRegionTest.java @@ -85,12 +85,15 @@ protected boolean shouldCreateD2Client() { public void setUp() { Properties serverProperties = new Properties(); serverProperties.putAll(getExtraServerProperties()); + serverProperties.putAll(ServiceFactory.getPubSubClientConfigs()); Properties parentControllerProps = new Properties(); parentControllerProps.put(ConfigKeys.CONTROLLER_AUTO_MATERIALIZE_META_SYSTEM_STORE, true); parentControllerProps.putAll(getExtraParentControllerProperties()); + parentControllerProps.putAll(ServiceFactory.getPubSubClientConfigs()); Properties childControllerProps = new Properties(); childControllerProps.put(ConfigKeys.CONTROLLER_AUTO_MATERIALIZE_META_SYSTEM_STORE, true); childControllerProps.putAll(getExtraChildControllerProperties()); + childControllerProps.putAll(ServiceFactory.getPubSubClientConfigs()); VeniceMultiRegionClusterCreateOptions.Builder optionsBuilder = new VeniceMultiRegionClusterCreateOptions.Builder().numberOfRegions(getNumberOfRegions()) .numberOfClusters(getNumberOfClusters()) diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/ActiveActiveReplicationForHybridTest.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/ActiveActiveReplicationForHybridTest.java index e9263aa2f4c..5aed7fdab05 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/ActiveActiveReplicationForHybridTest.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/ActiveActiveReplicationForHybridTest.java @@ -70,6 +70,7 @@ import org.apache.helix.manager.zk.ZKHelixAdmin; import org.apache.helix.model.IdealState; import org.apache.http.HttpStatus; +import org.apache.samza.config.MapConfig; import org.apache.samza.system.OutgoingMessageEnvelope; import org.apache.samza.system.SystemStream; import org.testng.annotations.AfterClass; @@ -469,6 +470,7 @@ public void testAAReplicationCanResolveConflicts(boolean useLogicalTimestamp, bo VeniceMultiClusterWrapper childDataCenter = childDatacenters.get(0); try (VeniceSystemProducer producerInDC0 = new VeniceSystemProducer( new VeniceSystemProducerConfig.Builder().setFactory(new VeniceSystemFactory()) + .setSamzaConfig(new MapConfig(childDataCenter.getPubSubClientProperties())) .setStoreName(storeName) .setPushType(Version.PushType.STREAM) .setSamzaJobId(Utils.getUniqueString("venice-push-id")) @@ -563,6 +565,7 @@ public void testAAReplicationCanResolveConflicts(boolean useLogicalTimestamp, bo VeniceMultiClusterWrapper childDataCenter1 = childDatacenters.get(1); try (VeniceSystemProducer producerInDC1 = new VeniceSystemProducer( new VeniceSystemProducerConfig.Builder().setFactory(new VeniceSystemFactory()) + .setSamzaConfig(new MapConfig(childDataCenter1.getPubSubClientProperties())) .setStoreName(storeName) .setPushType(Version.PushType.STREAM) .setSamzaJobId(Utils.getUniqueString("venice-push-id")) diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciClientP2PBlobTransferTest.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciClientP2PBlobTransferTest.java index 704eae29fd4..53c8b0b9829 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciClientP2PBlobTransferTest.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciClientP2PBlobTransferTest.java @@ -143,6 +143,7 @@ public void testBlobP2PTransferAmongDVC(boolean batchPushReportEnable, Boolean i File configDir = Utils.getTempDataDirectory(); File configFile = new File(configDir, "dvc-config.properties"); Properties props = new Properties(); + props.putAll(ServiceFactory.getPubSubClientConfigs()); props.setProperty("zk.hosts", zkHosts); props.setProperty("base.data.path", dvcPath1); props.setProperty("store.name", storeName); @@ -291,6 +292,7 @@ public void testBlobP2PTransferForNonLaggingDaVinciClient() throws Exception { File configDir = Utils.getTempDataDirectory(); File configFile = new File(configDir, "dvc-config.properties"); Properties props = new Properties(); + props.putAll(ServiceFactory.getPubSubClientConfigs()); props.setProperty("zk.hosts", zkHosts); props.setProperty("base.data.path", dvcPath1); props.setProperty("store.name", storeName); diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciClientRecordTransformerFilterTest.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciClientRecordTransformerFilterTest.java index e6d37c25922..72a844c7d37 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciClientRecordTransformerFilterTest.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciClientRecordTransformerFilterTest.java @@ -56,6 +56,7 @@ import com.linkedin.venice.controllerapi.UpdateStoreQueryParams; import com.linkedin.venice.integration.utils.DaVinciTestContext; import com.linkedin.venice.integration.utils.PubSubBrokerWrapper; +import com.linkedin.venice.integration.utils.ServiceFactory; import com.linkedin.venice.integration.utils.VeniceClusterWrapper; import com.linkedin.venice.integration.utils.VeniceRouterWrapper; import com.linkedin.venice.meta.Version; @@ -463,6 +464,7 @@ private List getDataMessages(String storeName, int keyCoun // Consume all the RT messages and validated how many data records were produced. PubSubBrokerWrapper pubSubBrokerWrapper = cluster.getPubSubBrokerWrapper(); Properties properties = new Properties(); + properties.putAll(ServiceFactory.getPubSubClientConfigs()); properties.setProperty(ConfigKeys.KAFKA_BOOTSTRAP_SERVERS, pubSubBrokerWrapper.getAddress()); List messages = new ArrayList<>(); diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciClientRecordTransformerTest.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciClientRecordTransformerTest.java index 6f6d10a44fc..5947bd39c19 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciClientRecordTransformerTest.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciClientRecordTransformerTest.java @@ -28,6 +28,7 @@ import com.linkedin.venice.controllerapi.ControllerClient; import com.linkedin.venice.controllerapi.UpdateStoreQueryParams; import com.linkedin.venice.integration.utils.DaVinciTestContext; +import com.linkedin.venice.integration.utils.ServiceFactory; import com.linkedin.venice.integration.utils.VeniceClusterWrapper; import com.linkedin.venice.integration.utils.VeniceRouterWrapper; import com.linkedin.venice.producer.online.OnlineProducerFactory; @@ -146,7 +147,7 @@ public void testRecordTransformer() throws Exception { ClientConfig.defaultGenericClientConfig(recordTransformerStoreName) .setD2Client(d2Client) .setD2ServiceName(VeniceRouterWrapper.CLUSTER_DISCOVERY_D2_SERVICE_NAME), - VeniceProperties.empty(), + new VeniceProperties(ServiceFactory.getPubSubClientConfigs()), null)) { producer.asyncDelete(1).get(); @@ -378,7 +379,7 @@ public void testRecordTransformerOnRecovery() throws Exception { ClientConfig.defaultGenericClientConfig(recordTransformerStoreName) .setD2Client(d2Client) .setD2ServiceName(VeniceRouterWrapper.CLUSTER_DISCOVERY_D2_SERVICE_NAME), - VeniceProperties.empty(), + new VeniceProperties(ServiceFactory.getPubSubClientConfigs()), null)) { int key = numKeys + 1; String value = "a" + key; diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciClusterAgnosticTest.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciClusterAgnosticTest.java index 278f5491f69..ef9db9831a9 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciClusterAgnosticTest.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciClusterAgnosticTest.java @@ -75,12 +75,14 @@ public class DaVinciClusterAgnosticTest { private VeniceMultiClusterWrapper multiClusterVenice; private String[] clusterNames; private String parentControllerURLs; + private Properties originalPubSubAdapterFactorySystemProperties; /** * Set up a multi-cluster Venice environment with meta system store enabled Venice stores. */ @BeforeClass public void setUp() { + originalPubSubAdapterFactorySystemProperties = ServiceFactory.setPubSubClientConfigsAsSystemProperties(); Utils.thisIsLocalhost(); Properties parentControllerProps = new Properties(); parentControllerProps.put(OFFLINE_JOB_START_TIMEOUT_MS, "180000"); @@ -109,9 +111,16 @@ public void setUp() { IntegrationTestUtils.waitForParticipantStorePush(clusterNames, multiClusterVenice.getControllerConnectString()); } - @AfterClass + @AfterClass(alwaysRun = true) public void cleanUp() { Utils.closeQuietlyWithErrorLogged(multiRegionMultiClusterWrapper); + restorePubSubAdapterFactorySystemProperties(); + } + + private void restorePubSubAdapterFactorySystemProperties() { + if (originalPubSubAdapterFactorySystemProperties != null) { + ServiceFactory.restorePubSubClientConfigsSystemProperties(originalPubSubAdapterFactorySystemProperties); + } } @Test(timeOut = 180 * Time.MS_PER_SECOND) diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciP2PBlobTransferRecoveryTest.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciP2PBlobTransferRecoveryTest.java index c21ffeea9eb..a8b53197555 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciP2PBlobTransferRecoveryTest.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciP2PBlobTransferRecoveryTest.java @@ -50,6 +50,7 @@ import com.linkedin.venice.compression.CompressionStrategy; import com.linkedin.venice.controllerapi.ControllerClient; import com.linkedin.venice.controllerapi.UpdateStoreQueryParams; +import com.linkedin.venice.integration.utils.ServiceFactory; import com.linkedin.venice.integration.utils.VeniceClusterWrapper; import com.linkedin.venice.integration.utils.VeniceRouterWrapper; import com.linkedin.venice.store.rocksdb.RocksDBUtils; @@ -120,6 +121,7 @@ public void testBlobP2PTransferAmongDVCWithServerShutdown(boolean isGracefulShut File configDir = Utils.getTempDataDirectory(); File configFile = new File(configDir, "dvc-config.properties"); Properties props = new Properties(); + props.putAll(ServiceFactory.getPubSubClientConfigs()); props.setProperty("zk.hosts", zkHosts); props.setProperty("base.data.path", dvcPath1); props.setProperty("store.name", storeName); @@ -265,6 +267,7 @@ public void testBlobP2PinDVCWithRestoreTempFolderSuccessfully() throws Exception File configDir = Utils.getTempDataDirectory(); File configFile = new File(configDir, "dvc-config.properties"); Properties props = new Properties(); + props.putAll(ServiceFactory.getPubSubClientConfigs()); props.setProperty("zk.hosts", zkHosts); props.setProperty("base.data.path", dvcPath1); props.setProperty("store.name", storeName); diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciP2PBlobTransferReportDisabledTest.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciP2PBlobTransferReportDisabledTest.java index e94dd565dc7..22ddcede91e 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciP2PBlobTransferReportDisabledTest.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciP2PBlobTransferReportDisabledTest.java @@ -45,6 +45,7 @@ import com.linkedin.venice.compression.CompressionStrategy; import com.linkedin.venice.controllerapi.ControllerClient; import com.linkedin.venice.controllerapi.UpdateStoreQueryParams; +import com.linkedin.venice.integration.utils.ServiceFactory; import com.linkedin.venice.integration.utils.VeniceClusterWrapper; import com.linkedin.venice.integration.utils.VeniceRouterWrapper; import com.linkedin.venice.store.rocksdb.RocksDBUtils; @@ -109,6 +110,7 @@ public void testBlobP2PTransferAmongDVC() throws Exception { File configDir = Utils.getTempDataDirectory(); File configFile = new File(configDir, "dvc-config.properties"); Properties props = new Properties(); + props.putAll(ServiceFactory.getPubSubClientConfigs()); props.setProperty("zk.hosts", zkHosts); props.setProperty("base.data.path", dvcPath1); props.setProperty("store.name", storeName); diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/OnlineVeniceProducerOrderingIntegrationTest.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/OnlineVeniceProducerOrderingIntegrationTest.java index e04eec2717f..9103b390787 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/OnlineVeniceProducerOrderingIntegrationTest.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/OnlineVeniceProducerOrderingIntegrationTest.java @@ -186,7 +186,8 @@ private VeniceProducer createConfiguredProducer( int callbackThreadCount, int callbackQueueCapacity, MetricsRepository metricsRepository) { - VeniceProperties producerConfig = new PropertyBuilder().put(CLIENT_PRODUCER_WORKER_COUNT, workerCount) + VeniceProperties producerConfig = new PropertyBuilder().put(ServiceFactory.getPubSubClientConfigs()) + .put(CLIENT_PRODUCER_WORKER_COUNT, workerCount) .put(CLIENT_PRODUCER_WORKER_QUEUE_CAPACITY, workerQueueCapacity) .put(CLIENT_PRODUCER_CALLBACK_THREAD_COUNT, callbackThreadCount) .put(CLIENT_PRODUCER_CALLBACK_QUEUE_CAPACITY, callbackQueueCapacity) diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/PushStatusStoreTest.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/PushStatusStoreTest.java index d61625eb257..4b1c23ec9fe 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/PushStatusStoreTest.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/PushStatusStoreTest.java @@ -94,6 +94,7 @@ public class PushStatusStoreTest { @BeforeClass public void setUp() { Properties extraProperties = new Properties(); + extraProperties.putAll(ServiceFactory.getPubSubClientConfigs()); // all tests in this class will be reading incremental push status from push status store extraProperties.setProperty(USE_PUSH_STATUS_STORE_FOR_INCREMENTAL_PUSH, String.valueOf(true)); extraProperties.setProperty( diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/StoreMetadataRecoveryTest.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/StoreMetadataRecoveryTest.java index 6f641c6b06f..4892cb47277 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/StoreMetadataRecoveryTest.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/StoreMetadataRecoveryTest.java @@ -45,9 +45,11 @@ public class StoreMetadataRecoveryTest { private String parentZKUrl; private String parentKafkaUrl; private String childControllerUrl; + private Properties originalPubSubAdapterFactorySystemProperties; @BeforeClass public void setUp() { + originalPubSubAdapterFactorySystemProperties = ServiceFactory.setPubSubClientConfigsAsSystemProperties(); Utils.thisIsLocalhost(); Properties parentControllerProperties = new Properties(); // Disable topic cleanup since parent and child are sharing the same kafka cluster. @@ -88,6 +90,13 @@ public void setUp() { @AfterClass(alwaysRun = true) public void cleanUp() { Utils.closeQuietlyWithErrorLogged(twoLayerClusterWrapper); + restorePubSubAdapterFactorySystemProperties(); + } + + private void restorePubSubAdapterFactorySystemProperties() { + if (originalPubSubAdapterFactorySystemProperties != null) { + ServiceFactory.restorePubSubClientConfigsSystemProperties(originalPubSubAdapterFactorySystemProperties); + } } @Test(timeOut = TEST_TIMEOUT) diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestActiveActiveReplicationWithDownRegion.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestActiveActiveReplicationWithDownRegion.java index 86d18947132..804ddb75907 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestActiveActiveReplicationWithDownRegion.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestActiveActiveReplicationWithDownRegion.java @@ -27,6 +27,7 @@ import java.util.concurrent.TimeUnit; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; +import org.apache.samza.config.MapConfig; import org.apache.samza.system.OutgoingMessageEnvelope; import org.apache.samza.system.SystemProducer; import org.apache.samza.system.SystemStream; @@ -111,6 +112,7 @@ public void testDownedKafka() throws Exception { // Build a system producer that writes nearline to dc-0 SystemProducer producerInDC0 = new VeniceSystemProducer( new VeniceSystemProducerConfig.Builder().setFactory(new VeniceSystemFactory()) + .setSamzaConfig(new MapConfig(childDatacenters.get(0).getPubSubClientProperties())) .setStoreName(storeName) .setPushType(Version.PushType.STREAM) .setSamzaJobId(Utils.getUniqueString("venice-push-id")) @@ -124,6 +126,7 @@ public void testDownedKafka() throws Exception { SystemProducer producerInDC1 = new VeniceSystemProducer( new VeniceSystemProducerConfig.Builder().setFactory(new VeniceSystemFactory()) + .setSamzaConfig(new MapConfig(childDatacenters.get(1).getPubSubClientProperties())) .setStoreName(storeName) .setPushType(Version.PushType.STREAM) .setSamzaJobId(Utils.getUniqueString("venice-push-id")) @@ -138,6 +141,7 @@ public void testDownedKafka() throws Exception { // Build another one which will write some batch data SystemProducer batchProducer = new VeniceSystemProducer( new VeniceSystemProducerConfig.Builder().setFactory(new VeniceSystemFactory()) + .setSamzaConfig(new MapConfig(childDatacenters.get(0).getPubSubClientProperties())) .setStoreName(storeName) .setPushType(Version.PushType.BATCH) .setSamzaJobId(Utils.getUniqueString("venice-push-id")) diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestAdminOperationWithPreviousVersion.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestAdminOperationWithPreviousVersion.java index ee96be2ccc0..7669f4702bd 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestAdminOperationWithPreviousVersion.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestAdminOperationWithPreviousVersion.java @@ -216,9 +216,11 @@ public class TestAdminOperationWithPreviousVersion { private Admin veniceAdmin; private List childControllerClients; private VeniceMultiClusterWrapper multiClusterWrapperRegion0; + private Properties originalPubSubAdapterFactorySystemProperties; @BeforeClass(alwaysRun = true) public void setUp() throws Exception { + originalPubSubAdapterFactorySystemProperties = ServiceFactory.setPubSubClientConfigsAsSystemProperties(); Utils.thisIsLocalhost(); // Validate that all operations have test coverage BEFORE running any tests @@ -294,6 +296,13 @@ void beforeEachTest() { @AfterClass(alwaysRun = true) public void cleanUp() { multiRegionMultiClusterWrapper.close(); + restorePubSubAdapterFactorySystemProperties(); + } + + private void restorePubSubAdapterFactorySystemProperties() { + if (originalPubSubAdapterFactorySystemProperties != null) { + ServiceFactory.restorePubSubClientConfigsSystemProperties(originalPubSubAdapterFactorySystemProperties); + } } /** diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestBatch.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestBatch.java index 9cda85d71b7..fa397b8939d 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestBatch.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestBatch.java @@ -59,6 +59,7 @@ import com.linkedin.venice.controllerapi.UpdateStoreQueryParams; import com.linkedin.venice.controllerapi.VersionCreationResponse; import com.linkedin.venice.exceptions.VeniceException; +import com.linkedin.venice.integration.utils.ServiceFactory; import com.linkedin.venice.integration.utils.VeniceClusterWrapper; import com.linkedin.venice.integration.utils.VeniceRouterWrapper; import com.linkedin.venice.jobs.StageMetricsSnapshot; @@ -455,6 +456,7 @@ public void testNewPushWithNewerDictionaryIsServedCorrectly() throws Exception { // Verify that v1 and v2 have different dictionaries (different data produces different dictionaries) Properties props = new Properties(); props.setProperty(KAFKA_BOOTSTRAP_SERVERS, veniceCluster.getPubSubBrokerWrapper().getAddress()); + props.putAll(ServiceFactory.getPubSubClientConfigs()); VeniceProperties veniceProperties = new VeniceProperties(props); ByteBuffer v1Dict = DictionaryUtils.readDictionaryFromKafka(Version.composeKafkaTopic(storeName, 1), veniceProperties); diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestBatchForRocksDB.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestBatchForRocksDB.java index 7457acc434f..4e82f9bd85c 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestBatchForRocksDB.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestBatchForRocksDB.java @@ -31,6 +31,7 @@ public VeniceClusterWrapper initializeVeniceCluster() { VeniceClusterWrapper veniceClusterWrapper = ServiceFactory.getVeniceCluster(options); Properties serverProperties = new Properties(); + serverProperties.putAll(ServiceFactory.getPubSubClientConfigs()); serverProperties.put(PERSISTENCE_TYPE, PersistenceType.ROCKS_DB); serverProperties.setProperty(ROCKSDB_PLAIN_TABLE_FORMAT_ENABLED, "false"); serverProperties.setProperty(SERVER_DATABASE_CHECKSUM_VERIFICATION_ENABLED, "true"); diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestDeferredVersionSwapDvc.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestDeferredVersionSwapDvc.java index a2f3853051e..a9a30a3a28b 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestDeferredVersionSwapDvc.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestDeferredVersionSwapDvc.java @@ -75,9 +75,11 @@ public class TestDeferredVersionSwapDvc { IntStream.range(0, NUMBER_OF_CLUSTERS).mapToObj(i -> "venice-cluster" + i).toArray(String[]::new); private static final int TEST_TIMEOUT = 120_000; private List childDatacenters; + private Properties originalPubSubAdapterFactorySystemProperties; @BeforeClass public void setUp() { + originalPubSubAdapterFactorySystemProperties = ServiceFactory.setPubSubClientConfigsAsSystemProperties(); Properties controllerProps = new Properties(); controllerProps.put(CONTROLLER_DEFERRED_VERSION_SWAP_SLEEP_MS, 100); controllerProps.put(CONTROLLER_DEFERRED_VERSION_SWAP_SERVICE_ENABLED, true); @@ -103,6 +105,13 @@ public void setUp() { @AfterClass(alwaysRun = true) public void cleanUp() { Utils.closeQuietlyWithErrorLogged(multiRegionMultiClusterWrapper); + restorePubSubAdapterFactorySystemProperties(); + } + + private void restorePubSubAdapterFactorySystemProperties() { + if (originalPubSubAdapterFactorySystemProperties != null) { + ServiceFactory.restorePubSubClientConfigsSystemProperties(originalPubSubAdapterFactorySystemProperties); + } } @Test(timeOut = TEST_TIMEOUT * 2) diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestDumpIngestionContext.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestDumpIngestionContext.java index b9a9c63b48b..2157d86c5a1 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestDumpIngestionContext.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestDumpIngestionContext.java @@ -14,6 +14,7 @@ import com.linkedin.venice.controllerapi.UpdateStoreQueryParams; import com.linkedin.venice.controllerapi.VersionCreationResponse; import com.linkedin.venice.exceptions.VeniceException; +import com.linkedin.venice.integration.utils.ServiceFactory; import com.linkedin.venice.integration.utils.VeniceClusterWrapper; import com.linkedin.venice.integration.utils.VeniceServerWrapper; import com.linkedin.venice.meta.Store; @@ -21,17 +22,33 @@ import com.linkedin.venice.utils.TestUtils; import com.linkedin.venice.utils.Utils; import java.util.Map; +import java.util.Properties; import java.util.concurrent.TimeUnit; import org.apache.avro.Schema; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.testng.Assert; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; import org.testng.annotations.Test; public class TestDumpIngestionContext extends AbstractMultiRegionTest { private static final Logger LOGGER = LogManager.getLogger(TestDumpIngestionContext.class); private static final int TEST_TIMEOUT_MS = 180_000; + private Properties originalPubSubAdapterFactorySystemProperties; + + @BeforeClass(alwaysRun = true) + public void setUpAdminToolSystemProperties() { + originalPubSubAdapterFactorySystemProperties = ServiceFactory.setPubSubClientConfigsAsSystemProperties(); + } + + @AfterClass(alwaysRun = true) + public void restoreAdminToolSystemProperties() { + if (originalPubSubAdapterFactorySystemProperties != null) { + ServiceFactory.restorePubSubClientConfigsSystemProperties(originalPubSubAdapterFactorySystemProperties); + } + } @Test(timeOut = TEST_TIMEOUT_MS) public void testDumpHostHeartbeatLag() { diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestHybrid.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestHybrid.java index ce402b309c1..21af82b4110 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestHybrid.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestHybrid.java @@ -1020,7 +1020,7 @@ public void testHybridMultipleVersions() throws Exception { try (VeniceProducer veniceOnlineProducer = OnlineProducerFactory.createProducer( ClientConfig.defaultGenericClientConfig(storeName).setVeniceURL(cluster.getRandomRouterURL()), - VeniceProperties.empty(), + new VeniceProperties(ServiceFactory.getPubSubClientConfigs()), null)) { for (int i = keyCount; i < keyCount * 2; i++) { veniceOnlineProducer.asyncPut(i, i * 2).get(); diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestPushJobWithNativeReplication.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestPushJobWithNativeReplication.java index 8e1b7c8258e..5cf102fe544 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestPushJobWithNativeReplication.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestPushJobWithNativeReplication.java @@ -104,6 +104,7 @@ protected boolean shouldCreateD2Client() { @Override protected Properties getExtraServerProperties() { Properties serverProperties = new Properties(); + serverProperties.putAll(ServiceFactory.getPubSubClientConfigs()); serverProperties.setProperty(SERVER_DATABASE_SYNC_BYTES_INTERNAL_FOR_DEFERRED_WRITE_MODE, "300"); return serverProperties; } @@ -111,6 +112,7 @@ protected Properties getExtraServerProperties() { @Override protected Properties getExtraControllerProperties() { Properties controllerProps = new Properties(); + controllerProps.putAll(ServiceFactory.getPubSubClientConfigs()); // This property is required for test stores that have 10 partitions controllerProps.put(DEFAULT_MAX_NUMBER_OF_PARTITIONS, 10); controllerProps.put(BatchJobHeartbeatConfigs.HEARTBEAT_STORE_CLUSTER_CONFIG.getConfigName(), SYSTEM_STORE_CLUSTER); diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestRepushCore.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestRepushCore.java index 6d5bd413644..0ff8dba7032 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestRepushCore.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestRepushCore.java @@ -46,6 +46,7 @@ import com.linkedin.venice.controllerapi.VersionCreationResponse; import com.linkedin.venice.exceptions.VeniceException; import com.linkedin.venice.hadoop.VenicePushJob; +import com.linkedin.venice.integration.utils.ServiceFactory; import com.linkedin.venice.integration.utils.VeniceClusterWrapper; import com.linkedin.venice.integration.utils.VeniceMultiClusterWrapper; import com.linkedin.venice.integration.utils.VeniceRouterWrapper; @@ -307,6 +308,7 @@ public void testRepushWithDeleteRecord(boolean useSpark) { .put(ROCKSDB_PLAIN_TABLE_FORMAT_ENABLED, "false") .put(ROCKSDB_BLOCK_CACHE_SIZE_IN_BYTES, 2 * 1024 * 1024L) .put(DAVINCI_PUSH_STATUS_CHECK_INTERVAL_IN_MS, 1000) + .put(ServiceFactory.getPubSubClientConfigs()) .build(); MetricsRepository metricsRepository = new VeniceMetricsRepository(); diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestStoreMigration.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestStoreMigration.java index 0993e2c499f..423d6250564 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestStoreMigration.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestStoreMigration.java @@ -129,9 +129,11 @@ public class TestStoreMigration { private String parentControllerUrl; private String childControllerUrl0; protected Client r2Client; + private Properties originalPubSubAdapterFactorySystemProperties; @BeforeClass public void setUp() throws Exception { + originalPubSubAdapterFactorySystemProperties = ServiceFactory.setPubSubClientConfigsAsSystemProperties(); Utils.thisIsLocalhost(); Properties parentControllerProperties = new Properties(); // Disable topic cleanup since parent and child are sharing the same kafka cluster. @@ -179,6 +181,13 @@ public void setUp() throws Exception { @AfterClass(alwaysRun = true) public void cleanUp() { Utils.closeQuietlyWithErrorLogged(twoLayerMultiRegionMultiClusterWrapper); + restorePubSubAdapterFactorySystemProperties(); + } + + private void restorePubSubAdapterFactorySystemProperties() { + if (originalPubSubAdapterFactorySystemProperties != null) { + ServiceFactory.restorePubSubClientConfigsSystemProperties(originalPubSubAdapterFactorySystemProperties); + } } @Test(timeOut = TEST_TIMEOUT) diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestStoreMigrationMultiRegion.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestStoreMigrationMultiRegion.java index 6e2e65a397d..140dd33f38c 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestStoreMigrationMultiRegion.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestStoreMigrationMultiRegion.java @@ -66,9 +66,11 @@ public class TestStoreMigrationMultiRegion { private String parentControllerUrl; private String childControllerUrl0; private String childControllerUrl1; + private Properties originalPubSubAdapterFactorySystemProperties; @BeforeClass(timeOut = 180_000) public void setUp() { + originalPubSubAdapterFactorySystemProperties = ServiceFactory.setPubSubClientConfigsAsSystemProperties(); Utils.thisIsLocalhost(); Properties controllerProperties = new Properties(); controllerProperties.setProperty(TOPIC_CLEANUP_SLEEP_INTERVAL_BETWEEN_TOPIC_LIST_FETCH_MS, String.valueOf(4000)); @@ -111,6 +113,13 @@ public void setUp() { @AfterClass(alwaysRun = true) public void cleanUp() { Utils.closeQuietlyWithErrorLogged(twoLayerMultiRegionMultiClusterWrapper); + restorePubSubAdapterFactorySystemProperties(); + } + + private void restorePubSubAdapterFactorySystemProperties() { + if (originalPubSubAdapterFactorySystemProperties != null) { + ServiceFactory.restorePubSubClientConfigsSystemProperties(originalPubSubAdapterFactorySystemProperties); + } } @Test(timeOut = TEST_TIMEOUT) diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestVTConsistencyCheckerJob.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestVTConsistencyCheckerJob.java index fd60fad2463..59484a30cb9 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestVTConsistencyCheckerJob.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestVTConsistencyCheckerJob.java @@ -24,6 +24,7 @@ import com.linkedin.venice.controllerapi.UpdateStoreQueryParams; import com.linkedin.venice.exceptions.VeniceException; import com.linkedin.venice.integration.utils.PubSubBrokerWrapper; +import com.linkedin.venice.integration.utils.ServiceFactory; import com.linkedin.venice.integration.utils.VeniceClusterWrapper; import com.linkedin.venice.meta.Version; import com.linkedin.venice.pubsub.adapter.kafka.common.ApacheKafkaOffsetPosition; @@ -48,6 +49,7 @@ import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; import org.apache.avro.Schema; +import org.apache.samza.config.MapConfig; import org.apache.samza.system.OutgoingMessageEnvelope; import org.apache.samza.system.SystemStream; import org.apache.spark.sql.Dataset; @@ -143,6 +145,7 @@ public void testFullPipelineWithBatchPushRTWritesAndInjectedInconsistency() thro // 2. RT writes from both DCs via Samza VeniceSystemProducer producerInDC0 = new VeniceSystemProducer( new VeniceSystemProducerConfig.Builder().setFactory(new VeniceSystemFactory()) + .setSamzaConfig(new MapConfig(childDatacenters.get(0).getPubSubClientProperties())) .setStoreName(storeName) .setPushType(Version.PushType.STREAM) .setSamzaJobId(Utils.getUniqueString("venice-push-id")) @@ -162,6 +165,7 @@ public void testFullPipelineWithBatchPushRTWritesAndInjectedInconsistency() thro VeniceSystemProducer producerInDC1 = new VeniceSystemProducer( new VeniceSystemProducerConfig.Builder().setFactory(new VeniceSystemFactory()) + .setSamzaConfig(new MapConfig(childDatacenters.get(1).getPubSubClientProperties())) .setStoreName(storeName) .setPushType(Version.PushType.STREAM) .setSamzaJobId(Utils.getUniqueString("venice-push-id")) @@ -205,6 +209,7 @@ public void testFullPipelineWithBatchPushRTWritesAndInjectedInconsistency() thro // 5. Send more RT writes after injection to advance HW and make the scenario more realistic VeniceSystemProducer postInjectionProducer = new VeniceSystemProducer( new VeniceSystemProducerConfig.Builder().setFactory(new VeniceSystemFactory()) + .setSamzaConfig(new MapConfig(childDatacenters.get(0).getPubSubClientProperties())) .setStoreName(storeName) .setPushType(Version.PushType.STREAM) .setSamzaJobId(Utils.getUniqueString("venice-push-id")) @@ -235,6 +240,7 @@ public void testFullPipelineWithBatchPushRTWritesAndInjectedInconsistency() thro File outputDir = new File(tempRoot, "output"); try { Properties jobProps = new Properties(); + jobProps.putAll(ServiceFactory.getPubSubClientConfigs()); jobProps.setProperty( VTConsistencyCheckerJob.DC0_BROKER_URL, childDatacenters.get(0).getPubSubBrokerWrapper().getAddress()); diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/VersionSpecificDaVinciClientTest.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/VersionSpecificDaVinciClientTest.java index 97ad6639598..01e6029940c 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/VersionSpecificDaVinciClientTest.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/VersionSpecificDaVinciClientTest.java @@ -127,7 +127,7 @@ public void testVersionSpecificDaVinciClient() throws Exception { ClientConfig.defaultGenericClientConfig(storeName) .setD2Client(d2Client) .setD2ServiceName(VeniceRouterWrapper.CLUSTER_DISCOVERY_D2_SERVICE_NAME), - VeniceProperties.empty(), + new VeniceProperties(ServiceFactory.getPubSubClientConfigs()), null)) { producer.asyncPut(streamingKey1, customValue).get(); diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/fastclient/utils/AbstractClientEndToEndSetup.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/fastclient/utils/AbstractClientEndToEndSetup.java index f0f4beafe92..1ba8560f2d6 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/fastclient/utils/AbstractClientEndToEndSetup.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/fastclient/utils/AbstractClientEndToEndSetup.java @@ -170,6 +170,7 @@ protected Properties getExtraServerProperties() { public void setUp() throws Exception { Utils.thisIsLocalhost(); Properties props = new Properties(); + props.putAll(ServiceFactory.getPubSubClientConfigs()); props.put(SERVER_HTTP2_INBOUND_ENABLED, "true"); props.put(SERVER_QUOTA_ENFORCEMENT_ENABLED, "true"); props.putAll(getExtraServerProperties()); diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/hadoop/input/kafka/TestKafkaInputFormat.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/hadoop/input/kafka/TestKafkaInputFormat.java index 75740f63074..0f2d9b0f6da 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/hadoop/input/kafka/TestKafkaInputFormat.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/hadoop/input/kafka/TestKafkaInputFormat.java @@ -136,6 +136,7 @@ public void testGetSplits() { KafkaInputFormat kafkaInputFormat = new KafkaInputFormat(); PubSubTopic topic = getTopic(1000, 3); JobConf conf = new JobConf(); + ServiceFactory.getPubSubClientConfigs().forEach((key, value) -> conf.set(key.toString(), value.toString())); conf.set(VENICE_REPUSH_SOURCE_PUBSUB_BROKER, pubSubBrokerWrapper.getAddress()); conf.set(KAFKA_INPUT_TOPIC, topic.getName()); diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/integration/utils/DaVinciTestContext.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/integration/utils/DaVinciTestContext.java index 508a7e2ec3d..19985bc7ad7 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/integration/utils/DaVinciTestContext.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/integration/utils/DaVinciTestContext.java @@ -197,6 +197,7 @@ public static PropertyBuilder getDaVinciPropertyBuilder(String zkAddress) { .put(CLIENT_SYSTEM_STORE_REPOSITORY_REFRESH_INTERVAL_SECONDS, 1) .put(D2_ZK_HOSTS_ADDRESS, zkAddress) .put(ROCKSDB_BLOCK_CACHE_SIZE_IN_BYTES, 4 * 1024 * 1024 * 1024L) - .put(CLUSTER_DISCOVERY_D2_SERVICE, VeniceRouterWrapper.CLUSTER_DISCOVERY_D2_SERVICE_NAME); + .put(CLUSTER_DISCOVERY_D2_SERVICE, VeniceRouterWrapper.CLUSTER_DISCOVERY_D2_SERVICE_NAME) + .put(ServiceFactory.getPubSubClientConfigs()); } } diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/integration/utils/KafkaBrokerFactory.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/integration/utils/KafkaBrokerFactory.java index 3a00eb22228..afbf479d63c 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/integration/utils/KafkaBrokerFactory.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/integration/utils/KafkaBrokerFactory.java @@ -17,7 +17,6 @@ import com.linkedin.venice.utils.VeniceProperties; import java.io.File; import java.util.ArrayList; -import java.util.Collections; import java.util.HashMap; import java.util.Map; import java.util.Properties; @@ -286,9 +285,25 @@ public String getPubSubClusterName() { @Override public Map getAdditionalConfig() { - return Collections.singletonMap( + Map configs = new HashMap<>(); + configs.put( ConfigKeys.PUBSUB_TYPE_ID_TO_POSITION_CLASS_NAME_MAP, VeniceProperties.mapToString(PubSubPositionTypeRegistry.RESERVED_POSITION_TYPE_ID_TO_CLASS_NAME_MAP)); + // Explicitly advertise the Apache Kafka adapter factories so that clients relying on + // getBrokerDetailsForClients() do not depend on the (now disabled by default) implicit Kafka fallback. + configs.put( + ConfigKeys.PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS, + KAFKA_CLIENTS_FACTORY.getProducerAdapterFactory().getClass().getName()); + configs.put( + ConfigKeys.PUBSUB_CONSUMER_ADAPTER_FACTORY_CLASS, + KAFKA_CLIENTS_FACTORY.getConsumerAdapterFactory().getClass().getName()); + configs.put( + ConfigKeys.PUBSUB_ADMIN_ADAPTER_FACTORY_CLASS, + KAFKA_CLIENTS_FACTORY.getAdminAdapterFactory().getClass().getName()); + configs.put( + ConfigKeys.PUBSUB_SOURCE_OF_TRUTH_ADMIN_ADAPTER_FACTORY_CLASS, + KAFKA_CLIENTS_FACTORY.getAdminAdapterFactory().getClass().getName()); + return configs; } @Override diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/integration/utils/ServiceFactory.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/integration/utils/ServiceFactory.java index 47a13bf3287..3698f318e40 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/integration/utils/ServiceFactory.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/integration/utils/ServiceFactory.java @@ -3,6 +3,10 @@ import static com.linkedin.venice.ConfigKeys.CLIENT_USE_SYSTEM_STORE_REPOSITORY; import static com.linkedin.venice.ConfigKeys.D2_ZK_HOSTS_ADDRESS; import static com.linkedin.venice.ConfigKeys.DATA_BASE_PATH; +import static com.linkedin.venice.ConfigKeys.PUBSUB_ADMIN_ADAPTER_FACTORY_CLASS; +import static com.linkedin.venice.ConfigKeys.PUBSUB_CONSUMER_ADAPTER_FACTORY_CLASS; +import static com.linkedin.venice.ConfigKeys.PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS; +import static com.linkedin.venice.ConfigKeys.PUBSUB_SOURCE_OF_TRUTH_ADMIN_ADAPTER_FACTORY_CLASS; import static com.linkedin.venice.integration.utils.VeniceClusterWrapperConstants.DEFAULT_MAX_ATTEMPT; import static com.linkedin.venice.integration.utils.VeniceClusterWrapperConstants.DEFAULT_WAIT_TIME_FOR_CLUSTER_START_S; import static com.linkedin.venice.integration.utils.VeniceClusterWrapperConstants.STANDALONE_REGION_NAME; @@ -120,10 +124,67 @@ public static void withMaxAttempt(int maxAttempt, Runnable action) { /** * @return an instance of {@link PubSubClientsFactory} */ - static PubSubClientsFactory getPubSubClientsFactory() { + public static PubSubClientsFactory getPubSubClientsFactory() { return PUBSUB_BROKER_FACTORY.getClientsFactory(); } + /** + * Resolves the pub-sub adapter factory class configs from the configured pub-sub backend (selected via the + * {@code pubSubBrokerFactory} system property) instead of hard-coding a specific implementation. Tests use this so + * the suite exercises whatever client configs the pub-sub backend under test exposes. + * + * @return a {@link Properties} carrying the producer, consumer, admin, and source-of-truth-admin adapter factory + * class names exposed by the configured backend. + */ + public static Properties getPubSubClientConfigs() { + PubSubClientsFactory clientsFactory = getPubSubClientsFactory(); + String adminAdapterFactoryClass = clientsFactory.getAdminAdapterFactory().getClass().getName(); + Properties properties = new Properties(); + properties.setProperty( + PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS, + clientsFactory.getProducerAdapterFactory().getClass().getName()); + properties.setProperty( + PUBSUB_CONSUMER_ADAPTER_FACTORY_CLASS, + clientsFactory.getConsumerAdapterFactory().getClass().getName()); + properties.setProperty(PUBSUB_ADMIN_ADAPTER_FACTORY_CLASS, adminAdapterFactoryClass); + properties.setProperty(PUBSUB_SOURCE_OF_TRUTH_ADMIN_ADAPTER_FACTORY_CLASS, adminAdapterFactoryClass); + return properties; + } + + /** + * Publishes the configured backend's pub-sub adapter factory class configs as JVM system properties so components + * that read from {@link System#getProperties()} (e.g. the admin tool) pick them up. + * + * @return the prior values of the affected system properties, to be passed to + * {@link #restorePubSubClientConfigsSystemProperties(Properties)} for cleanup. + */ + public static Properties setPubSubClientConfigsAsSystemProperties() { + Properties originalProperties = new Properties(); + Properties factoryConfigs = getPubSubClientConfigs(); + for (String key: factoryConfigs.stringPropertyNames()) { + String originalValue = System.getProperty(key); + if (originalValue != null) { + originalProperties.setProperty(key, originalValue); + } + System.setProperty(key, factoryConfigs.getProperty(key)); + } + return originalProperties; + } + + /** + * Restores the system properties previously mutated by {@link #setPubSubClientConfigsAsSystemProperties()}. + */ + public static void restorePubSubClientConfigsSystemProperties(Properties originalProperties) { + for (String key: getPubSubClientConfigs().stringPropertyNames()) { + if (originalProperties.containsKey(key)) { + System.setProperty(key, originalProperties.getProperty(key)); + } else { + System.clearProperty(key); + } + } + originalProperties.clear(); + } + /** * @return an instance of {@link ZkServerWrapper} */ diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/utils/IntegrationTestPushUtils.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/utils/IntegrationTestPushUtils.java index 4ab816ca3fe..3ad9c1f9def 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/utils/IntegrationTestPushUtils.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/utils/IntegrationTestPushUtils.java @@ -51,6 +51,7 @@ import com.linkedin.venice.helix.VeniceJsonSerializer; import com.linkedin.venice.integration.utils.KafkaTestUtils; import com.linkedin.venice.integration.utils.PubSubBrokerWrapper; +import com.linkedin.venice.integration.utils.ServiceFactory; import com.linkedin.venice.integration.utils.VeniceClusterWrapper; import com.linkedin.venice.integration.utils.VeniceControllerWrapper; import com.linkedin.venice.integration.utils.VeniceMultiClusterWrapper; @@ -286,6 +287,7 @@ public static Map getSamzaProducerConfig( samzaConfig.put(VENICE_PARENT_CONTROLLER_D2_SERVICE, PARENT_D2_SERVICE_NAME); samzaConfig.put(DEPLOYMENT_ID, Utils.getUniqueString("venice-push-id")); samzaConfig.put(SSL_ENABLED, "false"); + addPubSubApacheKafkaAdapterFactoryConfigs(samzaConfig); samzaConfig.putAll( PubSubBrokerWrapper.getBrokerDetailsForClients(Collections.singletonList(venice.getPubSubBrokerWrapper()))); return samzaConfig; @@ -306,6 +308,7 @@ private static Map getSamzaProducerConfig( samzaConfig.put(VENICE_PARENT_CONTROLLER_D2_SERVICE, PARENT_D2_SERVICE_NAME); samzaConfig.put(DEPLOYMENT_ID, "DC_" + index + "_" + storeName); samzaConfig.put(SSL_ENABLED, "false"); + samzaConfig.putAll(clusterWrapper.getChildRegions().get(index).getPubSubClientProperties()); return samzaConfig; } @@ -324,9 +327,14 @@ private static Map getSamzaProducerConfigForBatch( samzaConfig.put(DEPLOYMENT_ID, Utils.getUniqueString("venice-push-id")); samzaConfig.put(SSL_ENABLED, "false"); samzaConfig.put(configPrefix + VENICE_AGGREGATE, "true"); + samzaConfig.putAll(clusterWrapper.getChildRegions().get(0).getPubSubClientProperties()); return samzaConfig; } + private static void addPubSubApacheKafkaAdapterFactoryConfigs(Map config) { + ServiceFactory.getPubSubClientConfigs().forEach((key, value) -> config.put(key.toString(), value.toString())); + } + /** * Create Samza Producer in Single-Region setup with optional configs. */ @@ -602,6 +610,7 @@ public static VeniceWriterFactory getVeniceWriterFactory( PubSubBrokerWrapper pubSubBrokerWrapper, PubSubProducerAdapterFactory pubSubProducerAdapterFactory) { Properties veniceWriterProperties = new Properties(); + veniceWriterProperties.putAll(ServiceFactory.getPubSubClientConfigs()); veniceWriterProperties.put(KAFKA_BOOTSTRAP_SERVERS, pubSubBrokerWrapper.getAddress()); veniceWriterProperties .putAll(PubSubBrokerWrapper.getBrokerDetailsForClients(Collections.singletonList(pubSubBrokerWrapper))); diff --git a/internal/venice-test-common/src/main/java/com/linkedin/venice/utils/TestUtils.java b/internal/venice-test-common/src/main/java/com/linkedin/venice/utils/TestUtils.java index 432921d2bf7..a4e332bfcca 100644 --- a/internal/venice-test-common/src/main/java/com/linkedin/venice/utils/TestUtils.java +++ b/internal/venice-test-common/src/main/java/com/linkedin/venice/utils/TestUtils.java @@ -79,6 +79,9 @@ import com.linkedin.venice.pubsub.PubSubPositionTypeRegistry; import com.linkedin.venice.pubsub.PubSubProducerAdapterFactory; import com.linkedin.venice.pubsub.PubSubTopicRepository; +import com.linkedin.venice.pubsub.adapter.kafka.admin.ApacheKafkaAdminAdapterFactory; +import com.linkedin.venice.pubsub.adapter.kafka.consumer.ApacheKafkaConsumerAdapterFactory; +import com.linkedin.venice.pubsub.adapter.kafka.producer.ApacheKafkaProducerAdapterFactory; import com.linkedin.venice.pubsub.api.PubSubPosition; import com.linkedin.venice.pubsub.api.PubSubTopicType; import com.linkedin.venice.pubsub.manager.TopicManagerRepository; @@ -738,8 +741,33 @@ public static VeniceControllerMultiClusterConfig getMultiClusterConfigFromOneClu return new VeniceControllerMultiClusterConfig(configMap); } + /** + * Returns the Apache Kafka pub-sub adapter factory-class configs (producer, consumer, admin). + *

+ * Tests that build a {@link VeniceServerConfig} or {@link VeniceControllerClusterConfig} (which + * eagerly construct a {@code PubSubClientsFactory}) must supply these now that the implicit Apache + * Kafka fallback is disabled by default. See + * {@code ConfigKeys#PUBSUB_ADAPTER_FACTORY_KAFKA_FALLBACK_ENABLED}. + */ + public static Properties getPubSubApacheKafkaAdapterFactoryConfigs() { + Properties properties = new Properties(); + properties.setProperty( + ConfigKeys.PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS, + ApacheKafkaProducerAdapterFactory.class.getName()); + properties.setProperty( + ConfigKeys.PUBSUB_CONSUMER_ADAPTER_FACTORY_CLASS, + ApacheKafkaConsumerAdapterFactory.class.getName()); + properties + .setProperty(ConfigKeys.PUBSUB_ADMIN_ADAPTER_FACTORY_CLASS, ApacheKafkaAdminAdapterFactory.class.getName()); + properties.setProperty( + ConfigKeys.PUBSUB_SOURCE_OF_TRUTH_ADMIN_ADAPTER_FACTORY_CLASS, + ApacheKafkaAdminAdapterFactory.class.getName()); + return properties; + } + public static Properties getPropertiesForControllerConfig() { Properties properties = new Properties(); + properties.putAll(getPubSubApacheKafkaAdapterFactoryConfigs()); properties.put(ConfigKeys.CLUSTER_NAME, "test-cluster"); properties.put(ConfigKeys.CONTROLLER_NAME, "venice-controller"); properties.put(ConfigKeys.DEFAULT_REPLICA_FACTOR, "1"); diff --git a/internal/venice-test-common/src/main/java/com/linkedin/venice/utils/TestWriteUtils.java b/internal/venice-test-common/src/main/java/com/linkedin/venice/utils/TestWriteUtils.java index 61571b7ba9f..aea4e862715 100644 --- a/internal/venice-test-common/src/main/java/com/linkedin/venice/utils/TestWriteUtils.java +++ b/internal/venice-test-common/src/main/java/com/linkedin/venice/utils/TestWriteUtils.java @@ -1,6 +1,8 @@ package com.linkedin.venice.utils; import static com.linkedin.venice.ConfigKeys.MULTI_REGION; +import static com.linkedin.venice.ConfigKeys.PASS_THROUGH_CONFIG_PREFIXES_LIST_KEY; +import static com.linkedin.venice.ConfigKeys.PUBSUB_CLIENT_CONFIG_PREFIX; import static com.linkedin.venice.vpj.VenicePushJobConstants.CONTROLLER_REQUEST_RETRY_ATTEMPTS; import static com.linkedin.venice.vpj.VenicePushJobConstants.D2_ZK_HOSTS_PREFIX; import static com.linkedin.venice.vpj.VenicePushJobConstants.DEFAULT_KEY_FIELD_PROP; @@ -987,6 +989,8 @@ public static Properties defaultVPJProps( } private static Properties defaultVPJPropsInternal(Properties props, String inputDirPath, String storeName) { + TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs().forEach(props::putIfAbsent); + props.putIfAbsent(PASS_THROUGH_CONFIG_PREFIXES_LIST_KEY, PUBSUB_CLIENT_CONFIG_PREFIX); props.put(VENICE_STORE_NAME_PROP, storeName); props.put(INPUT_PATH_PROP, inputDirPath); // No need for a big close timeout in tests. This is just to speed up discovery of certain regressions. diff --git a/tests/venice-pulsar-test/src/pulsarIntegrationTest/java/com/linkedin/venice/pulsar/sink/PulsarVeniceSinkTest.java b/tests/venice-pulsar-test/src/pulsarIntegrationTest/java/com/linkedin/venice/pulsar/sink/PulsarVeniceSinkTest.java index 3d96f49e7dc..be4e64a08e1 100644 --- a/tests/venice-pulsar-test/src/pulsarIntegrationTest/java/com/linkedin/venice/pulsar/sink/PulsarVeniceSinkTest.java +++ b/tests/venice-pulsar-test/src/pulsarIntegrationTest/java/com/linkedin/venice/pulsar/sink/PulsarVeniceSinkTest.java @@ -31,6 +31,15 @@ public class PulsarVeniceSinkTest { private static final Logger LOGGER = LogManager.getLogger(PulsarVeniceSinkTest.class); + // Apache Kafka pub-sub adapter factory classes passed to admin-tool JVM invocations run inside the + // venice-client container. The admin tool fails fast when these are not provided (no implicit default). + private static final String PUBSUB_ADAPTER_FACTORY_JVM_ARGS = "-Dpubsub.producer.adapter.factory.class=" + + "com.linkedin.venice.pubsub.adapter.kafka.producer.ApacheKafkaProducerAdapterFactory " + + "-Dpubsub.consumer.adapter.factory.class=" + + "com.linkedin.venice.pubsub.adapter.kafka.consumer.ApacheKafkaConsumerAdapterFactory " + + "-Dpubsub.admin.adapter.factory.class=" + + "com.linkedin.venice.pubsub.adapter.kafka.admin.ApacheKafkaAdminAdapterFactory"; + private DockerComposeContainer environment; // Generated by delombok @@ -158,8 +167,8 @@ public void testPulsarVeniceSink() throws Exception { // Wait for the store to become queryable before proceeding LOGGER.info("Waiting for Venice store to be ready"); - String readinessCmd = "java -jar " + jar + " --describe-store --url " + veniceControllerUrl + " --cluster " - + clusterName + " --store " + storeName; + String readinessCmd = "java " + PUBSUB_ADAPTER_FACTORY_JVM_ARGS + " -jar " + jar + " --describe-store --url " + + veniceControllerUrl + " --cluster " + clusterName + " --store " + storeName; Awaitility.await().atMost(30, TimeUnit.SECONDS).pollInterval(2, TimeUnit.SECONDS).untilAsserted(() -> { ExecResult res = execByService("venice-client", "bash", "-c", readinessCmd); String stdout = res.getStdout(); @@ -269,8 +278,8 @@ private void initVeniceStore(String veniceControllerUrl, String jar, String clus "venice-client", "bash", "-c", - "java -jar " + jar + " --empty-push --url " + veniceControllerUrl + " --cluster " + clusterName + " --store " - + storeName + " --push-id init --store-size 1000"); + "java " + PUBSUB_ADAPTER_FACTORY_JVM_ARGS + " -jar " + jar + " --empty-push --url " + veniceControllerUrl + + " --cluster " + clusterName + " --store " + storeName + " --push-id init --store-size 1000"); } private void updateVeniceStoreQuotas(String veniceControllerUrl, String jar, String clusterName, String storeName) @@ -279,15 +288,16 @@ private void updateVeniceStoreQuotas(String veniceControllerUrl, String jar, Str "venice-client", "bash", "-c", - "java -jar " + jar + " --update-store --url " + veniceControllerUrl + " --cluster " + clusterName + " --store " - + storeName + " --storage-quota -1 --incremental-push-enabled true"); + "java " + PUBSUB_ADAPTER_FACTORY_JVM_ARGS + " -jar " + jar + " --update-store --url " + veniceControllerUrl + + " --cluster " + clusterName + " --store " + storeName + + " --storage-quota -1 --incremental-push-enabled true"); execByServiceAsssertNoStdErr( "venice-client", "bash", "-c", - "java -jar " + jar + " --update-store --url " + veniceControllerUrl + " --cluster " + clusterName + " --store " - + storeName + " --read-quota 1000000"); + "java " + PUBSUB_ADAPTER_FACTORY_JVM_ARGS + " -jar " + jar + " --update-store --url " + veniceControllerUrl + + " --cluster " + clusterName + " --store " + storeName + " --read-quota 1000000"); } private void createVeniceStore( @@ -301,8 +311,9 @@ private void createVeniceStore( "venice-client", "bash", "-c", - "java -jar " + jar + " --new-store --url " + veniceControllerUrl + " --cluster " + clusterName + " --store " - + storeName + " --key-schema-file " + keyFile + " --value-schema-file " + valueFile); + "java " + PUBSUB_ADAPTER_FACTORY_JVM_ARGS + " -jar " + jar + " --new-store --url " + veniceControllerUrl + + " --cluster " + clusterName + " --store " + storeName + " --key-schema-file " + keyFile + + " --value-schema-file " + valueFile); } private void saveKeyValueSchemaFiles(String keyAsvc, String valueAsvc, String keyFile, String valueFile)