add protection to eager swap before consumer is fully subscribed

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

@@ -175,6 +175,10 @@ public CompletableFuture<Void> seekToEndOfPush(Set<Integer> partitions) {
     });
   }
 
+  public boolean subscribed() {
+    return isSubscribed.get();
+  }
+
   @Override
   protected CompletableFuture<Void> internalSeek(
       Set<Integer> partitions,

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

@@ -68,6 +68,7 @@
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -106,6 +107,8 @@ public class VeniceChangelogConsumerImpl<K, V> implements VeniceChangelogConsume
   protected final Map<Integer, Boolean> partitionToBootstrapState = new VeniceConcurrentHashMap<>();
   protected final long startTimestamp;
 
+  protected final AtomicBoolean isSubscribed = new AtomicBoolean(false);
+
   protected final RecordDeserializer<K> keyDeserializer;
   private final D2ControllerClient d2ControllerClient;
   private final RecordDeserializer<RecordChangeEvent> recordChangeDeserializer =
@@ -260,8 +263,9 @@ protected CompletableFuture<Void> internalSubscribe(Set<Integer> partitions, Pub
           pubSubConsumer.subscribe(topicPartition, OffsetRecord.LOWEST_OFFSET);
           currentVersionLastHeartbeat.put(topicPartition.getPartitionNumber(), System.currentTimeMillis());
         }
-        return null;
       }
+      isSubscribed.set(true);
+      return null;
     });
   }
 

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

@@ -32,6 +32,10 @@ class VersionSwapDataChangeListener<K, V> implements StoreDataChangedListener {
   @Override
   public void handleStoreChanged(Store store) {
     // store may be null as this is called by other repair tasks
+    if (!consumer.subscribed()) {
+      // skip this for now as the consumer hasn't even been set up yet
+      return;
+    }
     synchronized (this) {
       Set<Integer> partitions = new HashSet<>();
       try {

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

@@ -266,8 +266,7 @@ public void testAAIngestionWithStoreView() throws Exception {
           .setControllerD2ServiceName(D2_SERVICE_NAME)
           .setD2ServiceName(VeniceRouterWrapper.CLUSTER_DISCOVERY_D2_SERVICE_NAME)
           .setLocalD2ZkHosts(localZkServer.getAddress())
-          .setControllerRequestRetryCount(3)
-          .setVersionSwapDetectionIntervalTimeInMs(10L);
+          .setControllerRequestRetryCount(3);
       VeniceChangelogConsumerClientFactory veniceChangelogConsumerClientFactory =
           new VeniceChangelogConsumerClientFactory(globalChangelogClientConfig, metricsRepository);