check if reads are enabled before reading in CDC client (linkedin#2341)

When reads are disabled for a store, thin/fast clients cannot read it's data.
However, CDC client does not respect this. In this change, CDC client will check
the `readEnabled` flag of the store before returning the data.

Author: arjun4084346

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

@@ -34,6 +34,7 @@
 import com.linkedin.venice.compression.VeniceCompressor;
 import com.linkedin.venice.controllerapi.D2ControllerClient;
 import com.linkedin.venice.exceptions.InvalidVeniceSchemaException;
+import com.linkedin.venice.exceptions.StoreDisabledException;
 import com.linkedin.venice.exceptions.StoreVersionNotFoundException;
 import com.linkedin.venice.exceptions.VeniceException;
 import com.linkedin.venice.kafka.protocol.ControlMessage;
@@ -343,6 +344,13 @@ public VeniceChangelogConsumerImpl(
     LOGGER.info("Start a change log consumer client for store: {}", storeName);
   }
 
+  public void throwIfReadsDisabled() {
+    Store store = getStore();
+    if (!store.isEnableReads()) {
+      throw new StoreDisabledException(store.getName(), "read");
+    }
+  }
+
   // Unit test only and read only
   VersionSwapMessageState getVersionSwapMessageState() {
     return this.versionSwapMessageState;
@@ -364,6 +372,7 @@ public CompletableFuture<Void> subscribe(Set<Integer> partitions) {
   }
 
   protected CompletableFuture<Void> internalSubscribe(Set<Integer> partitions, PubSubTopic topic) {
+    throwIfReadsDisabled();
     return CompletableFuture.supplyAsync(() -> {
       try {
         for (int i = 0; i <= MAX_SUBSCRIBE_RETRIES; i++) {
@@ -822,6 +831,7 @@ protected Collection<PubSubMessage<K, ChangeEvent<V>, VeniceChangeCoordinate>> i
       long timeoutInMs,
       String topicSuffix,
       boolean includeControlMessage) {
+    throwIfReadsDisabled();
     Collection<PubSubMessage<K, ChangeEvent<V>, VeniceChangeCoordinate>> pubSubMessages = new ArrayList<>();
     Map<PubSubTopicPartition, List<DefaultPubSubMessage>> messagesMap;
     boolean lockAcquired = false;

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

@@ -16,6 +16,7 @@
 import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doCallRealMethod;
+import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
@@ -185,6 +186,7 @@ public void setUp() {
     when(mockRepository.getValueSchema(storeName, 1)).thenReturn(new SchemaEntry(1, valueSchema));
     when(store.getVersionOrThrow(Mockito.anyInt())).thenReturn(mockVersion);
     when(store.getVersion(Mockito.anyInt())).thenReturn(mockVersion);
+    when(store.isEnableReads()).thenReturn(true);
 
     mockPubSubConsumer = mock(PubSubConsumerAdapter.class);
     oldVersionTopic = pubSubTopicRepository.getTopic(Version.composeKafkaTopic(storeName, 1));
@@ -315,6 +317,7 @@ public void testAfterImageConsumerSeek() throws ExecutionException, InterruptedE
     when(store.getPartitionCount()).thenReturn(2);
     when(mockRepository.getStore(anyString())).thenReturn(store);
     when(store.getVersion(Mockito.anyInt())).thenReturn(mockVersion);
+    when(store.isEnableReads()).thenReturn(true);
     veniceChangelogConsumer.setStoreRepository(mockRepository);
 
     Assert.assertEquals(veniceChangelogConsumer.getPartitionCount(), 2);
@@ -923,6 +926,7 @@ public void testChunkingSuccess() throws NoSuchFieldException, IllegalAccessExce
   @Test
   public void testChunkingFailure() throws NoSuchFieldException, IllegalAccessException {
     VeniceChangelogConsumerImpl veniceChangelogConsumer = mock(VeniceChangelogConsumerImpl.class);
+    doNothing().when(veniceChangelogConsumer).throwIfReadsDisabled();
 
     Field changelogClientConfigField = VeniceChangelogConsumerImpl.class.getDeclaredField("changelogClientConfig");
     changelogClientConfigField.setAccessible(true);
@@ -1038,6 +1042,9 @@ public void testConcurrentPolls() throws ExecutionException, InterruptedExceptio
         PubSubMessageDeserializer.createDefaultDeserializer(),
         veniceChangelogConsumerClientFactory);
 
+    VeniceChangelogConsumerImpl<String, Utf8> spyConsumer = Mockito.spy(veniceChangelogConsumer);
+    doNothing().when(spyConsumer).throwIfReadsDisabled();
+
     /*
      * We make this test deterministic by making the first poll hold the lock longer than the second poll, to ensure
      * the second poll times out before it can acquire the lock. Thus, ensuring poll on the PubSubConsumer only gets
@@ -1058,15 +1065,15 @@ public void testConcurrentPolls() throws ExecutionException, InterruptedExceptio
 
     // First poll task - will acquire the lock and hold it
     Callable<Void> firstPollTask = () -> {
-      veniceChangelogConsumer.poll(pollTimeoutMs * 4); // Long timeout to ensure it gets the lock
+      spyConsumer.poll(pollTimeoutMs * 4); // Long timeout to ensure it gets the lock
       return null;
     };
 
     // Second poll task - will timeout waiting for the lock
     Callable<Void> secondPollTask = () -> {
       // Wait for first poll to start, then try to poll with short timeout
       firstPollStarted.await();
-      veniceChangelogConsumer.poll(pollTimeoutMs / 2); // Short timeout to ensure it times out
+      spyConsumer.poll(pollTimeoutMs / 2); // Short timeout to ensure it times out
       return null;
     };
 
@@ -1141,6 +1148,7 @@ public void testPollBeforeSubscribeCompletes() throws ExecutionException, Interr
     when(delayedMockRepository.getValueSchema(storeName, 1)).thenReturn(new SchemaEntry(1, valueSchema));
     when(store.getVersionOrThrow(Mockito.anyInt())).thenReturn(mockVersion);
     when(store.getVersion(Mockito.anyInt())).thenReturn(mockVersion);
+    when(store.isEnableReads()).thenReturn(true);
 
     CountDownLatch subscribeStarted = new CountDownLatch(1);
 
@@ -1240,6 +1248,7 @@ public void testVersionSwapByControlMessage() throws ExecutionException, Interru
     when(store.getVersion(1)).thenReturn(mockOldVersion);
     when(store.getVersionOrThrow(2)).thenReturn(mockNewVersion);
     when(store.getVersion(2)).thenReturn(mockNewVersion);
+    when(store.isEnableReads()).thenReturn(true);
     veniceChangeLogConsumer.setStoreRepository(mockRepository);
 
     // partition 0 has an irrelevant version swap and partition 1 has a relevant version swap

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

@@ -8,6 +8,7 @@
 import static com.linkedin.venice.ConfigKeys.KAFKA_LINGER_MS;
 import static com.linkedin.venice.ConfigKeys.SERVER_AA_WC_WORKLOAD_PARALLEL_PROCESSING_ENABLED;
 import static com.linkedin.venice.ConfigKeys.ZOOKEEPER_ADDRESS;
+import static com.linkedin.venice.integration.utils.IntegrationTestUtils.pollChangeEventsFromSpecificChangeCaptureConsumer;
 import static com.linkedin.venice.integration.utils.VeniceClusterWrapperConstants.DEFAULT_PARENT_DATA_CENTER_REGION_NAME;
 import static com.linkedin.venice.integration.utils.VeniceControllerWrapper.D2_SERVICE_NAME;
 import static com.linkedin.venice.stats.ClientType.CHANGE_DATA_CAPTURE_CLIENT;
@@ -59,6 +60,7 @@
 import com.linkedin.venice.controllerapi.MultiStoreTopicsResponse;
 import com.linkedin.venice.controllerapi.UpdateStoreQueryParams;
 import com.linkedin.venice.endToEnd.TestChangelogValue;
+import com.linkedin.venice.exceptions.StoreDisabledException;
 import com.linkedin.venice.integration.utils.IntegrationTestUtils;
 import com.linkedin.venice.integration.utils.PubSubBrokerWrapper;
 import com.linkedin.venice.integration.utils.ServiceFactory;
@@ -188,6 +190,130 @@ public void cleanUp() {
     TestView.resetCounters();
   }
 
+  @Test(timeOut = TEST_TIMEOUT, priority = 3)
+  public void testDisabledStoreVeniceChangelogConsumer() throws Exception {
+    File inputDir = getTempDataDirectory();
+    Schema recordSchema = TestWriteUtils.writeSimpleAvroFileWithStringToNameRecordV1Schema(inputDir);
+    String inputDirPath = "file://" + inputDir.getAbsolutePath();
+    String storeName = Utils.getUniqueString("store");
+    Properties props = TestWriteUtils.defaultVPJProps(
+        parentControllers.get(0).getControllerUrl(),
+        inputDirPath,
+        storeName,
+        clusterWrapper.getPubSubClientProperties());
+    String keySchemaStr = recordSchema.getField(DEFAULT_KEY_FIELD_PROP).schema().toString();
+    String valueSchemaStr = NAME_RECORD_V2_SCHEMA.toString();
+    UpdateStoreQueryParams storeParms = new UpdateStoreQueryParams().setActiveActiveReplicationEnabled(true)
+        .setHybridRewindSeconds(500)
+        .setHybridOffsetLagThreshold(8)
+        .setChunkingEnabled(true)
+        .setNativeReplicationEnabled(true)
+        .setPartitionCount(3);
+    MetricsRepository metricsRepository =
+        getVeniceMetricsRepository(CHANGE_DATA_CAPTURE_CLIENT, CONSUMER_METRIC_ENTITIES, true);
+    ControllerClient setupControllerClient =
+        createStoreForJob(clusterName, keySchemaStr, valueSchemaStr, props, storeParms);
+    TestUtils.assertCommand(
+        setupControllerClient
+            .retryableRequest(5, controllerClient1 -> setupControllerClient.updateStore(storeName, storeParms)));
+    // Registering real data schema as schema v2.
+    for (Schema schema: SCHEMA_HISTORY) {
+      TestUtils.assertCommand(
+          setupControllerClient.retryableRequest(
+              5,
+              controllerClient1 -> setupControllerClient.addValueSchema(storeName, schema.toString())),
+          "Failed to add schema: " + schema.toString() + " to store " + storeName);
+    }
+
+    IntegrationTestPushUtils.runVPJ(props);
+    ZkServerWrapper localZkServer = multiRegionMultiClusterWrapper.getChildRegions().get(0).getZkServerWrapper();
+    PubSubBrokerWrapper localKafka = multiRegionMultiClusterWrapper.getChildRegions().get(0).getPubSubBrokerWrapper();
+    Properties consumerProperties = new Properties();
+    String localKafkaUrl = localKafka.getAddress();
+    consumerProperties.put(KAFKA_BOOTSTRAP_SERVERS, localKafkaUrl);
+    consumerProperties.put(CLUSTER_NAME, clusterName);
+    consumerProperties.put(ZOOKEEPER_ADDRESS, localZkServer.getAddress());
+    consumerProperties.putAll(multiRegionMultiClusterWrapper.getPubSubClientProperties());
+    ChangelogClientConfig globalChangelogClientConfig =
+        new ChangelogClientConfig().setConsumerProperties(consumerProperties)
+            .setControllerD2ServiceName(D2_SERVICE_NAME)
+            .setD2ServiceName(VeniceRouterWrapper.CLUSTER_DISCOVERY_D2_SERVICE_NAME)
+            .setD2Client(IntegrationTestPushUtils.getD2Client(localZkServer.getAddress()))
+            .setLocalD2ZkHosts(localZkServer.getAddress())
+            .setControllerRequestRetryCount(3)
+            .setVersionSwapDetectionIntervalTimeInSeconds(1L)
+            .setSpecificValue(TestChangelogValue.class)
+            .setBootstrapFileSystemPath(Utils.getUniqueString(inputDirPath));
+    VeniceChangelogConsumerClientFactory veniceChangelogConsumerClientFactory =
+        new VeniceChangelogConsumerClientFactory(globalChangelogClientConfig, metricsRepository);
+    VeniceChangelogConsumer<Utf8, TestChangelogValue> specificChangelogConsumer =
+        veniceChangelogConsumerClientFactory.getChangelogConsumer(storeName, "0", TestChangelogValue.class);
+
+    TestUtils.assertCommand(
+        setupControllerClient.retryableRequest(
+            5,
+            controllerClient1 -> setupControllerClient
+                .updateStore(storeName, new UpdateStoreQueryParams().setEnableReads(false))));
+
+    // Wait for store update to propagate
+    TestUtils.waitForNonDeterministicAssertion(
+        globalChangelogClientConfig.getVersionSwapDetectionIntervalTimeInSeconds(),
+        TimeUnit.SECONDS,
+        () -> Assert.assertThrows(StoreDisabledException.class, () -> specificChangelogConsumer.subscribeAll().get()));
+
+    TestUtils.assertCommand(
+        setupControllerClient.retryableRequest(
+            5,
+            controllerClient1 -> setupControllerClient
+                .updateStore(storeName, new UpdateStoreQueryParams().setEnableReads(true))));
+
+    specificChangelogConsumer.subscribeAll().get();
+
+    Map<String, PubSubMessage<Utf8, ChangeEvent<TestChangelogValue>, VeniceChangeCoordinate>> polledChangeEventsMap =
+        new HashMap<>();
+    List<PubSubMessage<Utf8, ChangeEvent<TestChangelogValue>, VeniceChangeCoordinate>> polledChangeEventsList =
+        new ArrayList<>();
+
+    TestUtils.assertCommand(
+        setupControllerClient.retryableRequest(
+            5,
+            controllerClient1 -> setupControllerClient
+                .updateStore(storeName, new UpdateStoreQueryParams().setEnableReads(false))));
+
+    // Wait for store update to propagate
+    TestUtils.waitForNonDeterministicAssertion(
+        globalChangelogClientConfig.getVersionSwapDetectionIntervalTimeInSeconds(),
+        TimeUnit.SECONDS,
+        () -> Assert.assertThrows(
+            StoreDisabledException.class,
+            () -> pollChangeEventsFromSpecificChangeCaptureConsumer(
+                polledChangeEventsMap,
+                polledChangeEventsList,
+                specificChangelogConsumer)));
+
+    TestUtils.assertCommand(
+        setupControllerClient.retryableRequest(
+            5,
+            controllerClient1 -> setupControllerClient
+                .updateStore(storeName, new UpdateStoreQueryParams().setEnableReads(true))));
+
+    TestUtils.waitForNonDeterministicAssertion(120, TimeUnit.SECONDS, true, () -> {
+      pollChangeEventsFromSpecificChangeCaptureConsumer(
+          polledChangeEventsMap,
+          polledChangeEventsList,
+          specificChangelogConsumer);
+      Assert.assertEquals(polledChangeEventsList.size(), 100);
+      Assert.assertTrue(specificChangelogConsumer.isCaughtUp());
+    });
+
+    Assert.assertTrue(
+        polledChangeEventsMap.get(Integer.toString(1)).getValue().getCurrentValue() instanceof SpecificRecord);
+    TestChangelogValue value = new TestChangelogValue();
+    value.firstName = "first_name_1";
+    value.lastName = "last_name_1";
+    Assert.assertEquals(polledChangeEventsMap.get(Integer.toString(1)).getValue().getCurrentValue(), value);
+  }
+
   // This is a beefier test, so giving it a bit more time
   @Test(timeOut = TEST_TIMEOUT * 3, priority = 3)
   public void testVersionSwapInALoop() throws Exception {