[da-vinci][changelog] Fix NPE when starting clients with no view configured (linkedin#2473)

  - Add empty string validation for view names in client factories                                                                                                                                         
  - Normalize empty view names to null in changelog consumers
  - Add integration tests for batch stores without views

Author: minhmo1620

clients/da-vinci-client/src/main/java/com/linkedin/davinci/client/factory/CachingDaVinciClientFactory.java

@@ -470,7 +470,8 @@ protected synchronized DaVinciClient getClient(
       DaVinciClientConstructor clientConstructor,
       Class clientClass,
       boolean startClient) {
-    String internalStoreName = viewName == null ? storeName : VeniceView.getViewStoreName(storeName, viewName);
+    String internalStoreName =
+        (viewName == null || viewName.isEmpty()) ? storeName : VeniceView.getViewStoreName(storeName, viewName);
     if (closed) {
       throw new VeniceException("Unable to get a client from a closed factory, storeName=" + internalStoreName);
     }

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

@@ -148,7 +148,11 @@ public SchemaReader getSchemaReader() {
   }
 
   public ChangelogClientConfig<T> setViewName(String viewName) {
-    this.viewName = viewName;
+    if (viewName != null && !viewName.isEmpty()) {
+      this.viewName = viewName;
+    } else {
+      this.viewName = null;
+    }
     return this;
   }
 

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

@@ -168,6 +168,22 @@ public void testGetChangelogConsumerWithConsumerId()
     Assert.assertNotSame(consumer1, consumer2);
   }
 
+  @Test
+  public void testChangelogConsumerWithViewName() {
+    ChangelogClientConfig globalChangelogClientConfig = new ChangelogClientConfig();
+
+    // Default of the field is null
+    Assert.assertNull(globalChangelogClientConfig.getViewName());
+
+    // Setting view name should work as expected
+    globalChangelogClientConfig.setViewName(VIEW_NAME);
+    Assert.assertEquals(globalChangelogClientConfig.getViewName(), VIEW_NAME);
+
+    // reset view name to null through empty string
+    globalChangelogClientConfig.setViewName("");
+    Assert.assertNull(globalChangelogClientConfig.getViewName());
+  }
+
   private void setUpMockStoreResponse(D2ControllerClient mockControllerClient, String storeConsumer) {
     StoreResponse mockStoreResponse = Mockito.mock(StoreResponse.class);
     Mockito.when(mockStoreResponse.isError()).thenReturn(false);

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestMaterializedViewEndToEnd.java

@@ -87,9 +87,11 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
+import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.IntStream;
@@ -638,6 +640,59 @@ public void testBatchOnlyMaterializedViewDVCConsumerWithOnePush() throws Excepti
     testDVCConsumer(storeName, 100);
   }
 
+  @Test(timeOut = TEST_TIMEOUT)
+  public void testBatchOnlyNoViewWithStatelessCDCConsumer() throws Exception {
+    testBatchOnlyNoViewWithCDCConsumer(false);
+  }
+
+  @Test(timeOut = TEST_TIMEOUT)
+  public void testBatchOnlyNoViewWithStatefulCDCConsumer() throws Exception {
+    testBatchOnlyNoViewWithCDCConsumer(true);
+  }
+
+  @Test(timeOut = TEST_TIMEOUT)
+  public void testBatchOnlyNoViewWithDVCConsumer() throws Exception {
+    // Create a batch only store with no view and run batch push job with 100 records
+    File inputDir = getTempDataDirectory();
+    Schema recordSchema = TestWriteUtils.writeSimpleAvroFileWithStringToStringSchema(inputDir);
+    String inputDirPath = "file:" + inputDir.getAbsolutePath();
+    String storeName = Utils.getUniqueString("batchStore");
+
+    setupStoreWithNoView(inputDirPath, storeName, recordSchema);
+
+    // Verify DVC consumer when there is no view.
+    // Start a DVC client that's subscribed to partition 0, 1, 3 of the store. The DVC client should
+    // contain all data records.
+    D2Client d2Client = D2TestUtils
+        .getAndStartD2Client(multiRegionMultiClusterWrapper.getChildRegions().get(1).getZkServerWrapper().getAddress());
+
+    VeniceProperties backendConfig =
+        new PropertyBuilder().put(DATA_BASE_PATH, Utils.getTempDataDirectory().getAbsolutePath())
+            .put(PERSISTENCE_TYPE, PersistenceType.ROCKS_DB)
+            .put(CLIENT_USE_SYSTEM_STORE_REPOSITORY, true)
+            .put(ROCKSDB_BLOCK_CACHE_SIZE_IN_BYTES, 2 * 1024 * 1024L)
+            .put(CLIENT_SYSTEM_STORE_REPOSITORY_REFRESH_INTERVAL_SECONDS, 1)
+            .build();
+    try (CachingDaVinciClientFactory factory = getCachingDaVinciClientFactory(
+        d2Client,
+        VeniceRouterWrapper.CLUSTER_DISCOVERY_D2_SERVICE_NAME,
+        new MetricsRepository(),
+        backendConfig,
+        multiRegionMultiClusterWrapper)) {
+      DaVinciClient<String, Object> client = factory.getAndStartGenericAvroClient(storeName, new DaVinciConfig());
+      Set<Integer> partitions = new HashSet<>();
+      partitions.add(0);
+      partitions.add(1);
+      partitions.add(2);
+      client.subscribe(partitions).get();
+      for (int i = 1; i <= 100; i++) {
+        assertEquals(client.get(Integer.toString(i)).get().toString(), DEFAULT_USER_DATA_VALUE_PREFIX + i);
+      }
+    } finally {
+      D2ClientUtils.shutdownClient(d2Client);
+    }
+  }
+
   private void testBatchOnlyMaterializedViewCDCConsumer(boolean useStatefulConsumer) throws Exception {
     // Create a batch only store with materialized view and run batch push job with 100 records
     File inputDir = getTempDataDirectory();
@@ -663,6 +718,29 @@ private void testBatchOnlyMaterializedViewCDCConsumer(boolean useStatefulConsume
     }
   }
 
+  public void testBatchOnlyNoViewWithCDCConsumer(boolean isStatefulClient) throws Exception {
+    // Create a batch only store with no view and run batch push job with 100 records
+    File inputDir = getTempDataDirectory();
+    Schema recordSchema = TestWriteUtils.writeSimpleAvroFileWithStringToStringSchema(inputDir);
+    String inputDirPath = "file:" + inputDir.getAbsolutePath();
+    String storeName = Utils.getUniqueString("batchStore");
+
+    setupStoreWithNoView(inputDirPath, storeName, recordSchema);
+
+    // Setup D2Client and ChangelogClientConfig
+    D2Client d2Client = D2TestUtils
+        .getAndStartD2Client(multiRegionMultiClusterWrapper.getChildRegions().get(0).getZkServerWrapper().getAddress());
+
+    try {
+      ChangelogClientConfig changelogClientConfig = createChangelogClientConfig(d2Client, inputDirPath, "");
+
+      // Test CDC consumer
+      testCDCConsumer(storeName, changelogClientConfig, isStatefulClient);
+    } finally {
+      D2ClientUtils.shutdownClient(d2Client);
+    }
+  }
+
   private void testDVCConsumer(String storeName, int recordCount) throws Exception {
     // Start a DVC client that's subscribed to partition 0 of the store's materialized view. The DVC client should
     // contain all data records.
@@ -693,6 +771,25 @@ private void testDVCConsumer(String storeName, int recordCount) throws Exception
     }
   }
 
+  private void setupStoreWithNoView(String inputDirPath, String storeName, Schema recordSchema) {
+    Properties props = TestWriteUtils.defaultVPJProps(
+        parentControllers.get(0).getControllerUrl(),
+        inputDirPath,
+        storeName,
+        multiRegionMultiClusterWrapper.getPubSubClientProperties());
+    String keySchemaStr = recordSchema.getField(DEFAULT_KEY_FIELD_PROP).schema().toString();
+    String valueSchemaStr = recordSchema.getField(DEFAULT_VALUE_FIELD_PROP).schema().toString();
+    UpdateStoreQueryParams storeParms = new UpdateStoreQueryParams().setActiveActiveReplicationEnabled(false)
+        .setChunkingEnabled(true)
+        .setRmdChunkingEnabled(true)
+        .setNativeReplicationEnabled(true)
+        .setNativeReplicationSourceFabric(childDatacenters.get(0).getRegionName())
+        .setPartitionCount(3);
+
+    IntegrationTestPushUtils.createStoreForJob(clusterName, keySchemaStr, valueSchemaStr, props, storeParms).close();
+    IntegrationTestPushUtils.runVPJ(props);
+  }
+
   private void setupStoreAndMaterializedView(
       String inputDirPath,
       String storeName,