[protocol][fast-client][server][build][compat] External-storage hooks + MetadataResponseRecord v3 activation (linkedin#2830)

Author: ymuppala

build.gradle

@@ -331,11 +331,7 @@ subprojects {
       def versionOverrides = [
           // AdminOperation v100 stages degradedDatacenters on AddVersion.
           // Pinned to the current active version until the Java wiring lands in a follow-up PR.
-          project(':services:venice-controller').file('src/main/resources/avro/AdminOperation/v99', PathValidation.DIRECTORY),
-          // MetadataResponseRecord v3 stages externalStorageReadMode (mirrors StoreProperties.externalStorageReadMode
-          // from StoreMetaValue v44). Pinned to the current active version until SERVER_METADATA_RESPONSE is bumped
-          // and server populate + Fast Client read land in follow-up PRs.
-          project(':internal:venice-common').file('src/main/resources/avro/MetadataResponseRecord/v2', PathValidation.DIRECTORY)
+          project(':services:venice-controller').file('src/main/resources/avro/AdminOperation/v99', PathValidation.DIRECTORY)
       ]
 
       def schemaDirs = [sourceDir]

clients/da-vinci-client/src/main/java/com/linkedin/davinci/listener/response/MetadataResponse.java

@@ -56,6 +56,10 @@ public void setBatchGetLimit(int batchGetLimit) {
     responseRecord.setBatchGetLimit(batchGetLimit);
   }
 
+  public void setExternalStorageReadMode(int externalStorageReadMode) {
+    responseRecord.setExternalStorageReadMode(externalStorageReadMode);
+  }
+
   public ByteBuf getResponseBody() {
     return Unpooled.wrappedBuffer(serializedResponse());
   }

clients/venice-client/src/main/java/com/linkedin/venice/fastclient/DelegatingAvroStoreClient.java

@@ -6,10 +6,13 @@
 import com.linkedin.venice.client.store.AvroGenericStoreClient;
 import com.linkedin.venice.client.store.ComputeGenericRecord;
 import com.linkedin.venice.client.store.ComputeRequestBuilder;
+import com.linkedin.venice.client.store.listeners.StoreConfigChangeListener;
+import com.linkedin.venice.client.store.listeners.StoreVersionSwitchListener;
 import com.linkedin.venice.client.store.streaming.StreamingCallback;
 import com.linkedin.venice.compute.ComputeRequestWrapper;
 import com.linkedin.venice.fastclient.factory.ClientFactory;
 import com.linkedin.venice.schema.SchemaReader;
+import java.nio.ByteBuffer;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
@@ -144,4 +147,19 @@ public ComputeRequestBuilder<K> compute(
       long preRequestTimeInNS) throws VeniceClientException {
     return delegate.compute(stats, streamingStats, computeStoreClient, preRequestTimeInNS);
   }
+
+  @Override
+  public V decompressAndDeserialize(ByteBuffer rawValue, int version, K key) throws VeniceClientException {
+    return delegate.decompressAndDeserialize(rawValue, version, key);
+  }
+
+  @Override
+  public void registerVersionSwitchListener(StoreVersionSwitchListener listener) {
+    delegate.registerVersionSwitchListener(listener);
+  }
+
+  @Override
+  public void registerStoreConfigChangeListener(StoreConfigChangeListener listener) {
+    delegate.registerStoreConfigChangeListener(listener);
+  }
 }

clients/venice-client/src/main/java/com/linkedin/venice/fastclient/DispatchingAvroGenericStoreClient.java

@@ -12,6 +12,8 @@
 import com.linkedin.venice.client.stats.ClientStats;
 import com.linkedin.venice.client.store.AbstractAvroStoreClient;
 import com.linkedin.venice.client.store.ComputeGenericRecord;
+import com.linkedin.venice.client.store.listeners.StoreConfigChangeListener;
+import com.linkedin.venice.client.store.listeners.StoreVersionSwitchListener;
 import com.linkedin.venice.client.store.streaming.ComputeRecordStreamDecoder;
 import com.linkedin.venice.client.store.streaming.StreamingCallback;
 import com.linkedin.venice.client.store.streaming.TrackingStreamingCallback;
@@ -41,6 +43,7 @@
 import com.linkedin.venice.utils.LatencyUtils;
 import com.linkedin.venice.utils.RedundantExceptionFilter;
 import com.linkedin.venice.utils.concurrent.ChainedCompletableFuture;
+import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -792,4 +795,45 @@ public Schema getLatestValueSchema() {
   public SchemaReader getSchemaReader() {
     return metadata;
   }
+
+  /**
+   * Fast Client implementation of the external-storage re-entry seam. Reads the 4-byte BE writer-schema-id prefix
+   * from {@code rawValue}, resolves the per-version compressor from {@code metadata} (including any ZSTD dictionary
+   * cached on prior refreshes), decompresses the remainder, and runs the existing deserialization pipeline at the
+   * embedded schema id. Reuses {@link #getDataRecordDeserializer} + {@link #tryToDeserialize} so any future change
+   * to the in-band read path propagates here automatically.
+   */
+  @Override
+  public V decompressAndDeserialize(ByteBuffer rawValue, int version, K key) throws VeniceClientException {
+    if (rawValue == null) {
+      throw new IllegalArgumentException("rawValue must not be null");
+    }
+    if (rawValue.remaining() < Integer.BYTES) {
+      throw new IllegalArgumentException(
+          "rawValue must hold a 4-byte writer-schema-id prefix; got " + rawValue.remaining() + " bytes");
+    }
+    // Work on a duplicate so reading the schemaId prefix + decompressing does not advance the caller's buffer
+    // position. duplicate() shares the underlying bytes but gives us our own position/limit/mark.
+    ByteBuffer view = rawValue.duplicate();
+    int schemaId = view.getInt();
+    CompressionStrategy strategy = metadata.getCompressionStrategy(version);
+    VeniceCompressor compressor = metadata.getCompressor(strategy, version);
+    ByteBuffer decompressed;
+    try {
+      decompressed = compressor.decompress(view);
+    } catch (IOException e) {
+      throw new VeniceClientException("Failed to decompress value bytes for store: " + getStoreName(), e);
+    }
+    return tryToDeserialize(getDataRecordDeserializer(schemaId), decompressed, schemaId, key);
+  }
+
+  @Override
+  public void registerVersionSwitchListener(StoreVersionSwitchListener listener) {
+    metadata.registerVersionSwitchListener(listener);
+  }
+
+  @Override
+  public void registerStoreConfigChangeListener(StoreConfigChangeListener listener) {
+    metadata.registerStoreConfigChangeListener(listener);
+  }
 }

clients/venice-client/src/main/java/com/linkedin/venice/fastclient/factory/ClientFactory.java

@@ -32,6 +32,21 @@ public static <K, V> AvroGenericStoreClient<K, V> getAndStartGenericStoreClient(
     return getAndStartGenericStoreClient(storeMetadata, clientConfig);
   }
 
+  /**
+   * Builds the Fast Client wrapper chain without calling {@link AvroGenericStoreClient#start()}. Callers that need
+   * to register a {@code StoreVersionSwitchListener} or {@code StoreConfigChangeListener} before the first metadata
+   * refresh fires should use this entry point, register on the returned client, then call {@code start()}.
+   *
+   * <p>{@code register*Listener} only delivers transitions observed after the listener exists in the registry — a
+   * listener registered after {@code start()} returns will not see the initial {@code (-1 -> currentVersion)}
+   * transition committed by the first refresh. This factory plus {@code start()} from the caller is the supported
+   * way to observe that initial transition.
+   */
+  public static <K, V> AvroGenericStoreClient<K, V> getGenericStoreClient(ClientConfig clientConfig) {
+    StoreMetadata storeMetadata = constructStoreMetadataReader(clientConfig);
+    return buildGenericStoreClient(storeMetadata, clientConfig);
+  }
+
   public static <K, V extends SpecificRecord> AvroSpecificStoreClient<K, V> getAndStartSpecificStoreClient(
       ClientConfig clientConfig) {
     /**
@@ -63,6 +78,18 @@ private static StoreMetadata constructStoreMetadataReader(ClientConfig clientCon
   public static <K, V> AvroGenericStoreClient<K, V> getAndStartGenericStoreClient(
       StoreMetadata storeMetadata,
       ClientConfig clientConfig) {
+    AvroGenericStoreClient<K, V> client = buildGenericStoreClient(storeMetadata, clientConfig);
+    client.start();
+    return client;
+  }
+
+  /**
+   * Builds the generic Fast Client wrapper chain over an externally-supplied {@link StoreMetadata} without calling
+   * {@link AvroGenericStoreClient#start()}. See {@link #getGenericStoreClient(ClientConfig)} for the rationale.
+   */
+  private static <K, V> AvroGenericStoreClient<K, V> buildGenericStoreClient(
+      StoreMetadata storeMetadata,
+      ClientConfig clientConfig) {
     final DispatchingAvroGenericStoreClient<K, V> dispatchingStoreClient = clientConfig.isVsonStore()
         ? new DispatchingVsonStoreClient<>(storeMetadata, clientConfig)
         : new DispatchingAvroGenericStoreClient<>(storeMetadata, clientConfig);
@@ -92,8 +119,6 @@ public static <K, V> AvroGenericStoreClient<K, V> getAndStartGenericStoreClient(
     if (clientConfig.isDualReadEnabled()) {
       dualReadClient = new DualReadAvroGenericStoreClient<>(statsStoreClient, clientConfig);
     }
-    dualReadClient.start();
-
     return dualReadClient;
   }
 

clients/venice-client/src/main/java/com/linkedin/venice/fastclient/meta/AbstractStoreMetadata.java

@@ -1,6 +1,9 @@
 package com.linkedin.venice.fastclient.meta;
 
 import com.linkedin.venice.client.exceptions.VeniceClientException;
+import com.linkedin.venice.client.store.listeners.StoreConfigChangeListener;
+import com.linkedin.venice.client.store.listeners.StoreConfigSnapshot;
+import com.linkedin.venice.client.store.listeners.StoreVersionSwitchListener;
 import com.linkedin.venice.client.store.transport.TransportClientResponse;
 import com.linkedin.venice.compression.CompressionStrategy;
 import com.linkedin.venice.compression.CompressorFactory;
@@ -24,6 +27,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.atomic.AtomicLong;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -36,6 +40,9 @@ public abstract class AbstractStoreMetadata implements StoreMetadata {
   private final InstanceHealthMonitor instanceHealthMonitor;
   protected volatile AbstractClientRoutingStrategy routingStrategy;
   protected final String storeName;
+  private final CopyOnWriteArrayList<StoreVersionSwitchListener> versionSwitchListeners = new CopyOnWriteArrayList<>();
+  private final CopyOnWriteArrayList<StoreConfigChangeListener> storeConfigChangeListeners =
+      new CopyOnWriteArrayList<>();
 
   public AbstractStoreMetadata(ClientConfig clientConfig) {
     this.clientConfig = clientConfig;
@@ -250,4 +257,90 @@ public <K> void routeRequest(RequestContext requestContext, RecordSerializer<K>
     throw new VeniceClientException("Unknown request type: " + requestType);
   }
 
+  @Override
+  public void registerVersionSwitchListener(StoreVersionSwitchListener listener) {
+    if (listener == null) {
+      throw new IllegalArgumentException("StoreVersionSwitchListener must not be null");
+    }
+    versionSwitchListeners.addIfAbsent(listener);
+  }
+
+  /**
+   * Return a {@link Runnable} that will notify the registered version-switch listeners of the supplied
+   * {@code (previousVersion, newVersion)} transition.
+   *
+   * <p>Build the {@link Runnable} inside the synchronized region that observed the transition, and run it after
+   * the region has been exited so that listeners may safely call back into this metadata.
+   */
+  protected Runnable buildVersionSwitchCallback(int previousVersion, int newVersion) {
+    return () -> fireVersionSwitch(previousVersion, newVersion);
+  }
+
+  /**
+   * Notify the registered version-switch listeners of a transition. Each listener is invoked in registration order;
+   * exceptions are caught and logged so that one failing listener cannot prevent others from running.
+   */
+  protected void fireVersionSwitch(int previousVersion, int newVersion) {
+    for (StoreVersionSwitchListener listener: versionSwitchListeners) {
+      try {
+        listener.onVersionSwitch(previousVersion, newVersion);
+      } catch (Throwable t) {
+        LOGGER.error(
+            "Store {} version-switch listener {} threw on transition {} -> {}",
+            storeName,
+            listener.getClass().getName(),
+            previousVersion,
+            newVersion,
+            t);
+      }
+    }
+  }
+
+  @Override
+  public void registerStoreConfigChangeListener(StoreConfigChangeListener listener) {
+    if (listener == null) {
+      throw new IllegalArgumentException("StoreConfigChangeListener must not be null");
+    }
+    storeConfigChangeListeners.addIfAbsent(listener);
+  }
+
+  /**
+   * Return a {@link Runnable} that will notify the registered store-config-change listeners of the supplied
+   * {@code (previous, current)} transition.
+   */
+  protected Runnable buildStoreConfigChangeCallback(StoreConfigSnapshot previous, StoreConfigSnapshot current) {
+    return () -> fireStoreConfigChange(previous, current);
+  }
+
+  /**
+   * Notify the registered store-config-change listeners of a transition, but only if {@code previous} and
+   * {@code current} differ — so callers may invoke this on every refresh without diffing first. Each listener is
+   * invoked in registration order; exceptions are caught and logged so that one failing listener cannot prevent
+   * others from running.
+   *
+   * @param previous  prior snapshot, or {@code null} if no prior snapshot existed
+   * @param current   new snapshot; must not be {@code null}
+   * @throws IllegalArgumentException if {@code current} is {@code null}
+   */
+  protected void fireStoreConfigChange(StoreConfigSnapshot previous, StoreConfigSnapshot current) {
+    if (current == null) {
+      throw new IllegalArgumentException("current snapshot must not be null");
+    }
+    if (current.equals(previous)) {
+      return;
+    }
+    for (StoreConfigChangeListener listener: storeConfigChangeListeners) {
+      try {
+        listener.onStoreConfigChange(previous, current);
+      } catch (Throwable t) {
+        LOGGER.error(
+            "Store {} store-config-change listener {} threw on transition {} -> {}",
+            storeName,
+            listener.getClass().getName(),
+            previous,
+            current,
+            t);
+      }
+    }
+  }
 }