[controller] use Utils::getRealTimeTopicName in tests

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

@@ -227,8 +227,6 @@ public void testisReadyToServeAnnouncedWithRTLag() {
         100L,
         DataReplicationPolicy.NON_AGGREGATE,
         BufferReplayPolicy.REWIND_FROM_EOP);
-    Version mockVersion = new VersionImpl(STORE_NAME, 1, PUSH_JOB_ID);
-    mockVersion.setHybridStoreConfig(hybridStoreConfig);
 
     StorageService storageService = mock(StorageService.class);
     Store store = new ZKStore(
@@ -241,6 +239,8 @@ public void testisReadyToServeAnnouncedWithRTLag() {
         OfflinePushStrategy.WAIT_ALL_REPLICAS,
         1);
     store.setHybridStoreConfig(hybridStoreConfig);
+    Version mockVersion = new VersionImpl(STORE_NAME, 1, PUSH_JOB_ID);
+    mockVersion.setHybridStoreConfig(hybridStoreConfig);
     store.setVersions(Collections.singletonList(mockVersion));
 
     Properties kafkaConsumerProperties = new Properties();

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

@@ -556,7 +556,7 @@ public void testKafkaConsumerServiceResubscriptionConcurrency() throws Exception
     KafkaConsumerServiceDelegator delegator =
         new KafkaConsumerServiceDelegator(mockConfig, consumerServiceBuilder, isAAWCStoreFunc);
     PubSubTopicPartition realTimeTopicPartition =
-        new PubSubTopicPartitionImpl(TOPIC_REPOSITORY.getTopic(Version.composeRealTimeTopic(storeName)), 0);
+        new PubSubTopicPartitionImpl(TOPIC_REPOSITORY.getTopic(Utils.composeRealTimeTopic(storeName)), 0);
 
     CountDownLatch countDownLatch = new CountDownLatch(1);
     List<Thread> infiniteSubUnSubThreads = new ArrayList<>();

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

@@ -134,7 +134,7 @@ public void testReportIfCatchUpBaseTopicOffsetRouteWillNotMakePushTimeout() {
      * {@link StoreIngestionTask#reportIfCatchUpVersionTopicOffset(PartitionConsumptionState)}
      */
     doReturn(true).when(mockOffsetRecord).isEndOfPushReceived();
-    doReturn(Version.composeRealTimeTopic(storeName)).when(mockOffsetRecord).getLeaderTopic();
+    doReturn(Utils.getRealTimeTopicName(mockStore)).when(mockOffsetRecord).getLeaderTopic();
     /**
      * Return 0 as the max offset for VT and 1 as the overall consume progress, so reportIfCatchUpVersionTopicOffset()
      * will determine that base topic is caught up.

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

@@ -52,6 +52,7 @@
 import static org.mockito.ArgumentMatchers.argThat;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.ArgumentMatchers.isA;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
 import static org.mockito.Mockito.after;
 import static org.mockito.Mockito.atLeast;
 import static org.mockito.Mockito.atLeastOnce;
@@ -135,6 +136,7 @@
 import com.linkedin.venice.meta.ReadOnlySchemaRepository;
 import com.linkedin.venice.meta.ReadOnlyStoreRepository;
 import com.linkedin.venice.meta.Store;
+import com.linkedin.venice.meta.StoreInfo;
 import com.linkedin.venice.meta.Version;
 import com.linkedin.venice.meta.VersionImpl;
 import com.linkedin.venice.meta.VersionStatus;
@@ -374,6 +376,7 @@ public static Object[][] sortedInputAndAAConfigProvider() {
   private StorePartitionDataReceiver remoteConsumedDataReceiver;
 
   private static String storeNameWithoutVersionInfo;
+  private StoreInfo storeInfo;
   private String topic;
   private PubSubTopic pubSubTopic;
   private PubSubTopicPartition fooTopicPartition;
@@ -498,6 +501,9 @@ public void methodSetUp() throws Exception {
     localVeniceWriter = getVeniceWriter(new MockInMemoryProducerAdapter(inMemoryLocalKafkaBroker));
 
     mockStorageEngineRepository = mock(StorageEngineRepository.class);
+    storeInfo = mock(StoreInfo.class, RETURNS_DEEP_STUBS);
+    when(storeInfo.getHybridStoreConfig().getRealTimeTopicName())
+        .thenReturn(Utils.composeRealTimeTopic(storeNameWithoutVersionInfo));
 
     mockLogNotifier = mock(LogNotifier.class);
     mockNotifierProgress = new ArrayList<>();
@@ -2542,7 +2548,7 @@ public void testDelayedTransitionToOnlineInHybridMode(AAConfig aaConfig) throws
 
       localVeniceWriter.broadcastTopicSwitch(
           Collections.singletonList(inMemoryLocalKafkaBroker.getKafkaBootstrapServer()),
-          Version.composeRealTimeTopic(storeNameWithoutVersionInfo),
+          Utils.getRealTimeTopicName(storeInfo),
           System.currentTimeMillis() - TimeUnit.SECONDS.toMillis(10),
           Collections.emptyMap());
       for (int partition: ALL_PARTITIONS) {
@@ -2958,7 +2964,7 @@ public void testRecordsCanBeThrottledPerRegion() throws ExecutionException, Inte
     kafkaUrlToRecordsThrottler.put(inMemoryLocalKafkaBroker.getKafkaBootstrapServer(), localThrottler);
     kafkaUrlToRecordsThrottler.put(inMemoryRemoteKafkaBroker.getKafkaBootstrapServer(), remoteThrottler);
 
-    String rtTopic = Version.composeRealTimeTopic(storeNameWithoutVersionInfo);
+    String rtTopic = Utils.getRealTimeTopicName(storeInfo);
     PubSubTopic rtPubSubTopic = pubSubTopicRepository.getTopic(rtTopic);
     PubSubTopicPartition fooRtPartition = new PubSubTopicPartitionImpl(rtPubSubTopic, PARTITION_FOO);
 
@@ -3088,7 +3094,7 @@ public void testIsReadyToServe(NodeType nodeType, AAConfig aaConfig, DataReplica
         false,
         Optional.empty(),
         null);
-    String rtTopicName = Version.composeRealTimeTopic(mockStore.getName());
+    String rtTopicName = Utils.getRealTimeTopicName(mockStore);
     PubSubTopic rtTopic = pubSubTopicRepository.getTopic(rtTopicName);
     TopicSwitch topicSwitchWithSourceRealTimeTopic = new TopicSwitch();
     topicSwitchWithSourceRealTimeTopic.sourceKafkaServers = new ArrayList<>();
@@ -3316,7 +3322,7 @@ public void testActiveActiveStoreIsReadyToServe(HybridConfig hybridConfig, NodeT
       assertNotNull(activeActiveStoreIngestionTask.getIngestionBatchProcessor().getLockManager());
     }
 
-    String rtTopicName = Version.composeRealTimeTopic(mockStore.getName());
+    String rtTopicName = Utils.getRealTimeTopicName(mockStore);
     PubSubTopic rtTopic = pubSubTopicRepository.getTopic(rtTopicName);
     TopicSwitch topicSwitchWithMultipleSourceKafkaServers = new TopicSwitch();
     topicSwitchWithMultipleSourceKafkaServers.sourceKafkaServers = new ArrayList<>();
@@ -3715,7 +3721,7 @@ public void testProcessTopicSwitch(NodeType nodeType) {
     TopicSwitch topicSwitchWithRemoteRealTimeTopic = new TopicSwitch();
     topicSwitchWithRemoteRealTimeTopic.sourceKafkaServers = new ArrayList<>();
     topicSwitchWithRemoteRealTimeTopic.sourceKafkaServers.add(inMemoryRemoteKafkaBroker.getKafkaBootstrapServer());
-    topicSwitchWithRemoteRealTimeTopic.sourceTopicName = Version.composeRealTimeTopic(mockStore.getName());
+    topicSwitchWithRemoteRealTimeTopic.sourceTopicName = Utils.getRealTimeTopicName(mockStore);
     topicSwitchWithRemoteRealTimeTopic.rewindStartTimestamp = System.currentTimeMillis();
     ControlMessage controlMessage = new ControlMessage();
     controlMessage.controlMessageUnion = topicSwitchWithRemoteRealTimeTopic;
@@ -3978,11 +3984,11 @@ private Consumer<PubSubTopicPartitionOffset> getObserver(
 
   @Test
   public void testResubscribeAfterRoleChange() throws Exception {
-    PubSubTopic realTimeTopic =
-        pubSubTopicRepository.getTopic(Version.composeRealTimeTopic(storeNameWithoutVersionInfo));
+    String realTimeTopicName = Utils.composeRealTimeTopic(storeNameWithoutVersionInfo);
+    PubSubTopic realTimeTopic = pubSubTopicRepository.getTopic(realTimeTopicName);
     // Prepare both local and remote real-time topics
-    inMemoryLocalKafkaBroker.createTopic(Version.composeRealTimeTopic(storeNameWithoutVersionInfo), PARTITION_COUNT);
-    inMemoryRemoteKafkaBroker.createTopic(Version.composeRealTimeTopic(storeNameWithoutVersionInfo), PARTITION_COUNT);
+    inMemoryLocalKafkaBroker.createTopic(realTimeTopicName, PARTITION_COUNT);
+    inMemoryRemoteKafkaBroker.createTopic(realTimeTopicName, PARTITION_COUNT);
     mockStorageMetadataService = new InMemoryStorageMetadataService();
 
     AbstractStoragePartition mockStoragePartition = mock(AbstractStoragePartition.class);
@@ -3995,12 +4001,10 @@ public void testResubscribeAfterRoleChange() throws Exception {
         .getReplicationMetadata(ByteBuffer.wrap(deleteKeyFoo));
 
     VeniceWriter vtWriter = getVeniceWriter(topic, new MockInMemoryProducerAdapter(inMemoryLocalKafkaBroker));
-    VeniceWriter localRtWriter = getVeniceWriter(
-        Version.composeRealTimeTopic(storeNameWithoutVersionInfo),
-        new MockInMemoryProducerAdapter(inMemoryLocalKafkaBroker));
-    VeniceWriter remoteRtWriter = getVeniceWriter(
-        Version.composeRealTimeTopic(storeNameWithoutVersionInfo),
-        new MockInMemoryProducerAdapter(inMemoryRemoteKafkaBroker));
+    VeniceWriter localRtWriter =
+        getVeniceWriter(realTimeTopicName, new MockInMemoryProducerAdapter(inMemoryLocalKafkaBroker));
+    VeniceWriter remoteRtWriter =
+        getVeniceWriter(realTimeTopicName, new MockInMemoryProducerAdapter(inMemoryRemoteKafkaBroker));
     HybridStoreConfig hybridStoreConfig = new HybridStoreConfigImpl(
         100,
         100,
@@ -4056,7 +4060,7 @@ public void testResubscribeAfterRoleChange() throws Exception {
           vtWriter.broadcastEndOfPush(new HashMap<>());
           vtWriter.broadcastTopicSwitch(
               kafkaBootstrapServers,
-              Version.composeRealTimeTopic(storeNameWithoutVersionInfo),
+              Utils.composeRealTimeTopic(storeNameWithoutVersionInfo),
               System.currentTimeMillis() - TimeUnit.SECONDS.toMillis(10),
               new HashMap<>());
           storeIngestionTaskUnderTest.promoteToLeader(

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

@@ -2,6 +2,7 @@
 
 import static com.linkedin.venice.kafka.protocol.enums.MessageType.PUT;
 import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
@@ -25,13 +26,13 @@
 import com.linkedin.venice.kafka.protocol.enums.MessageType;
 import com.linkedin.venice.message.KafkaKey;
 import com.linkedin.venice.meta.StoreInfo;
-import com.linkedin.venice.meta.Version;
 import com.linkedin.venice.pubsub.ImmutablePubSubMessage;
 import com.linkedin.venice.pubsub.PubSubTopicPartitionImpl;
 import com.linkedin.venice.pubsub.PubSubTopicRepository;
 import com.linkedin.venice.pubsub.adapter.kafka.consumer.ApacheKafkaConsumerAdapter;
 import com.linkedin.venice.pubsub.api.PubSubMessage;
 import com.linkedin.venice.pubsub.api.PubSubTopicPartition;
+import com.linkedin.venice.utils.Utils;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -50,7 +51,7 @@ public class TestAdminToolConsumption {
   @Test
   void testAdminToolAdminMessageConsumption() {
     int assignedPartition = 0;
-    String topic = Version.composeRealTimeTopic(STORE_NAME);
+    String topic = Utils.composeRealTimeTopic(STORE_NAME);
     PubSubTopicPartition pubSubTopicPartition =
         new PubSubTopicPartitionImpl(pubSubTopicRepository.getTopic(topic), assignedPartition);
     int adminMessageNum = 10;
@@ -112,16 +113,17 @@ private List<PubSubMessage<KafkaKey, KafkaMessageEnvelope, Long>> prepareAdminPu
 
   @Test
   public void testAdminToolConsumption() {
-    String topic = Version.composeRealTimeTopic(STORE_NAME);
     ControllerClient controllerClient = mock(ControllerClient.class);
     SchemaResponse schemaResponse = mock(SchemaResponse.class);
     when(schemaResponse.getSchemaStr()).thenReturn(SCHEMA_STRING);
     when(controllerClient.getKeySchema(STORE_NAME)).thenReturn(schemaResponse);
     StoreResponse storeResponse = mock(StoreResponse.class);
-    StoreInfo storeInfo = mock(StoreInfo.class);
+    StoreInfo storeInfo = mock(StoreInfo.class, RETURNS_DEEP_STUBS);
     when(storeInfo.getPartitionCount()).thenReturn(2);
     when(controllerClient.getStore(STORE_NAME)).thenReturn(storeResponse);
     when(storeResponse.getStore()).thenReturn(storeInfo);
+    when(storeInfo.getHybridStoreConfig().getRealTimeTopicName()).thenReturn(Utils.composeRealTimeTopic(STORE_NAME));
+    String topic = storeInfo.getHybridStoreConfig().getRealTimeTopicName();
 
     int assignedPartition = 0;
     long startOffset = 0;

clients/venice-producer/src/test/java/com/linkedin/venice/producer/online/OnlineVeniceProducerTest.java

@@ -909,7 +909,7 @@ private void configureMockTransportClient(
         versionCreationResponse.setPartitionerClass(partitionerConfig.getPartitionerClass());
         versionCreationResponse.setPartitionerParams(partitionerConfig.getPartitionerParams());
         versionCreationResponse.setKafkaBootstrapServers("localhost:9092");
-        versionCreationResponse.setKafkaTopic(Version.composeRealTimeTopic(storeName));
+        versionCreationResponse.setKafkaTopic(Utils.getRealTimeTopicName(store));
         versionCreationResponse.setEnableSSL(false);
 
         return getTransportClientFuture(MAPPER.writeValueAsBytes(versionCreationResponse), delayInResponseMs);

clients/venice-push-job/src/test/java/com/linkedin/venice/heartbeat/TestPushJobHeartbeatSender.java

@@ -9,10 +9,10 @@
 import com.linkedin.venice.meta.PartitionerConfig;
 import com.linkedin.venice.meta.PartitionerConfigImpl;
 import com.linkedin.venice.meta.StoreInfo;
-import com.linkedin.venice.meta.Version;
 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.Utils;
 import com.linkedin.venice.utils.VeniceProperties;
 import java.util.Optional;
 import java.util.Properties;
@@ -32,8 +32,10 @@ public void testHeartbeatSenderCreation() {
     // Prepare controller client.
     ControllerClient controllerClient = mock(ControllerClient.class);
     StoreResponse storeResponse = mock(StoreResponse.class);
-    StoreInfo storeInfo = mock(StoreInfo.class);
+    StoreInfo storeInfo = mock(StoreInfo.class, RETURNS_DEEP_STUBS);
     PartitionerConfig partitionerConfig = new PartitionerConfigImpl();
+    when(storeInfo.getHybridStoreConfig().getRealTimeTopicName())
+        .thenReturn(Utils.composeRealTimeTopic(heartbeatStoreName));
     doReturn(1).when(storeInfo).getPartitionCount();
     doReturn(partitionerConfig).when(storeInfo).getPartitionerConfig();
     doReturn(storeInfo).when(storeResponse).getStore();
@@ -60,6 +62,6 @@ public void testHeartbeatSenderCreation() {
         (DefaultPushJobHeartbeatSender) pushJobHeartbeatSender;
     Assert.assertEquals(
         defaultPushJobHeartbeatSender.getVeniceWriter().getTopicName(),
-        Version.composeRealTimeTopic(heartbeatStoreName));
+        Utils.composeRealTimeTopic(heartbeatStoreName));
   }
 }

clients/venice-thin-client/src/test/java/com/linkedin/venice/client/schema/RouterBackedSchemaReaderTest.java

@@ -530,7 +530,7 @@ private AbstractAvroStoreClient getMockStoreClient(boolean updateEnabled, int de
     versionCreationResponse.setPartitionerClass(partitionerConfig.getPartitionerClass());
     versionCreationResponse.setPartitionerParams(partitionerConfig.getPartitionerParams());
     versionCreationResponse.setKafkaBootstrapServers("localhost:9092");
-    versionCreationResponse.setKafkaTopic(Version.composeRealTimeTopic(storeName));
+    versionCreationResponse.setKafkaTopic(Utils.getRealTimeTopicName(store));
     versionCreationResponse.setEnableSSL(false);
 
     CompletableFuture<byte[]> requestTopicFuture = mock(CompletableFuture.class);

internal/venice-common/src/test/java/com/linkedin/venice/store/StoreStateReaderTest.java

@@ -67,8 +67,6 @@ public void testStoreStateReaderAPIBadResponse() throws ExecutionException, Inte
   private ZKStore getStore() {
     int partitionCount = 10;
     PartitionerConfig partitionerConfig = new PartitionerConfigImpl();
-    Version version = new VersionImpl(storeName, 1, "test-job-id");
-    version.setPartitionCount(partitionCount);
 
     HybridStoreConfig hybridStoreConfig = new HybridStoreConfigImpl(
         1000,
@@ -92,6 +90,8 @@ private ZKStore getStore() {
         partitionerConfig,
         3);
     store.setPartitionCount(partitionCount);
+    Version version = new VersionImpl(storeName, 1, "test-job-id");
+    version.setPartitionCount(partitionCount);
     store.setVersions(Collections.singletonList(version));
 
     return store;

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/consumer/ConsumerIntegrationTest.java

@@ -11,7 +11,6 @@
 import com.linkedin.venice.integration.utils.ServiceFactory;
 import com.linkedin.venice.integration.utils.VeniceClusterWrapper;
 import com.linkedin.venice.kafka.protocol.KafkaMessageEnvelope;
-import com.linkedin.venice.meta.Version;
 import com.linkedin.venice.partitioner.DefaultVenicePartitioner;
 import com.linkedin.venice.partitioner.VenicePartitioner;
 import com.linkedin.venice.pubsub.adapter.kafka.producer.ApacheKafkaProducerAdapter;
@@ -118,14 +117,15 @@ void extraBeforeClassSetUp(VeniceClusterWrapper cluster, ControllerClient contro
   public void testSetUp() {
     store = Utils.getUniqueString("consumer_integ_test");
     version = 1;
-    topicName = Version.composeRealTimeTopic(store);
     cluster.getNewStore(store);
     long streamingRewindSeconds = 25L;
     long streamingMessageLag = 2L;
     controllerClient.updateStore(
         store,
         new UpdateStoreQueryParams().setHybridRewindSeconds(streamingRewindSeconds)
             .setHybridOffsetLagThreshold(streamingMessageLag));
+    topicName = Utils.getRealTimeTopicName(
+        cluster.getLeaderVeniceController().getVeniceAdmin().getStore(cluster.getClusterName(), store));
     controllerClient.emptyPush(store, "test_push", 1);
     TestUtils.waitForNonDeterministicAssertion(15, TimeUnit.SECONDS, () -> {
       StoreResponse freshStoreResponse = controllerClient.getStore(store);

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/AbstractTestVeniceHelixAdmin.java

@@ -29,7 +29,6 @@
 import com.linkedin.venice.integration.utils.ServiceFactory;
 import com.linkedin.venice.integration.utils.ZkServerWrapper;
 import com.linkedin.venice.meta.Store;
-import com.linkedin.venice.meta.Version;
 import com.linkedin.venice.pubsub.PubSubTopicRepository;
 import com.linkedin.venice.stats.HelixMessageChannelStats;
 import com.linkedin.venice.utils.HelixUtils;
@@ -271,6 +270,6 @@ private void verifyParticipantMessageStoreSetup() {
         TimeUnit.SECONDS,
         () -> Assert.assertEquals(
             veniceAdmin.getRealTimeTopic(clusterName, participantStoreName),
-            Version.composeRealTimeTopic(participantStoreName)));
+            Utils.composeRealTimeTopic(participantStoreName)));
   }
 }