[router][common] Use copy of delegate inside of ReadOnlyStore for metadata serialization

internal/venice-common/src/main/java/com/linkedin/venice/meta/ReadOnlyStore.java

@@ -3,6 +3,7 @@
 import com.linkedin.venice.common.VeniceSystemStoreType;
 import com.linkedin.venice.compression.CompressionStrategy;
 import com.linkedin.venice.exceptions.StoreVersionNotFoundException;
+import com.linkedin.venice.exceptions.VeniceException;
 import com.linkedin.venice.systemstore.schemas.DataRecoveryConfig;
 import com.linkedin.venice.systemstore.schemas.StoreETLConfig;
 import com.linkedin.venice.systemstore.schemas.StoreHybridConfig;
@@ -877,6 +878,22 @@ public ReadOnlyStore(Store delegate) {
     this.delegate = delegate;
   }
 
+  public ZKStore getDelegateCopy() {
+    if (this.getClass() != ReadOnlyStore.class) {
+      throw new VeniceException(
+          "getDelegateCopy() was called on a subclass of ReadOnlyStore: " + this.getClass().getName() + ". "
+              + "Subclasses may override methods like getName(), getVersions(), or getPartitionCount() in ways that "
+              + "getDelegateCopy() cannot preserve, since it copies only the underlying delegate. "
+              + "Override getDelegateCopy() in your subclass, or add an explicit branch in the caller to handle this type.");
+    }
+    if (!(this.delegate instanceof ZKStore)) {
+      throw new VeniceException(
+          "ReadOnlyStore delegate is not a ZKStore, cannot produce a copy for serialization. Actual type: "
+              + this.delegate.getClass().getName());
+    }
+    return new ZKStore(this.delegate);
+  }
+
   @Override
   public String getName() {
     return this.delegate.getName();

services/venice-router/src/main/java/com/linkedin/venice/router/MetaDataHandler.java

@@ -39,6 +39,7 @@
 import com.linkedin.venice.controllerapi.SchemaResponse;
 import com.linkedin.venice.controllerapi.VersionCreationResponse;
 import com.linkedin.venice.exceptions.ErrorType;
+import com.linkedin.venice.exceptions.VeniceException;
 import com.linkedin.venice.exceptions.VeniceHttpException;
 import com.linkedin.venice.exceptions.VeniceNoHelixResourceException;
 import com.linkedin.venice.helix.HelixCustomizedViewOfflinePushRepository;
@@ -47,12 +48,15 @@
 import com.linkedin.venice.helix.SystemStoreJSONSerializer;
 import com.linkedin.venice.meta.PartitionerConfig;
 import com.linkedin.venice.meta.ReadOnlySchemaRepository;
+import com.linkedin.venice.meta.ReadOnlyStore;
 import com.linkedin.venice.meta.ReadOnlyStoreConfigRepository;
 import com.linkedin.venice.meta.ReadOnlyStoreRepository;
+import com.linkedin.venice.meta.ReadOnlyViewStore;
 import com.linkedin.venice.meta.Store;
 import com.linkedin.venice.meta.StoreConfig;
 import com.linkedin.venice.meta.SystemStore;
 import com.linkedin.venice.meta.Version;
+import com.linkedin.venice.meta.ZKStore;
 import com.linkedin.venice.pushmonitor.ExecutionStatus;
 import com.linkedin.venice.pushmonitor.HybridStoreQuotaStatus;
 import com.linkedin.venice.pushstatushelper.PushStatusStoreReader;
@@ -850,8 +854,17 @@ private void handleStoreStateLookup(ChannelHandlerContext ctx, VenicePathParserH
     if (store instanceof SystemStore) {
       SystemStore systemStore = (SystemStore) store;
       body = SYSTEM_STORE_SERIALIZER.serialize(systemStore.getSerializableSystemStore(), null);
-    } else {
+    } else if (store instanceof ReadOnlyViewStore) {
+      throw new VeniceException(
+          "Unexpected ReadOnlyViewStore encountered for store state lookup of store: " + storeName);
+    } else if (store instanceof ReadOnlyStore) {
+      body = STORE_SERIALIZER.serialize(((ReadOnlyStore) store).getDelegateCopy(), null);
+    } else if (store instanceof ZKStore) {
       body = STORE_SERIALIZER.serialize(store, null);
+    } else {
+      throw new VeniceException(
+          "Unexpected store type encountered for store state lookup of store: " + storeName + ", type: "
+              + store.getClass().getName());
     }
     setupResponseAndFlush(OK, body, true, ctx);
   }

services/venice-router/src/test/java/com/linkedin/venice/router/TestMetaDataHandler.java

@@ -45,6 +45,7 @@
 import com.linkedin.venice.meta.PartitionerConfig;
 import com.linkedin.venice.meta.PartitionerConfigImpl;
 import com.linkedin.venice.meta.ReadOnlySchemaRepository;
+import com.linkedin.venice.meta.ReadOnlyStore;
 import com.linkedin.venice.meta.SerializableSystemStore;
 import com.linkedin.venice.meta.Store;
 import com.linkedin.venice.meta.StoreConfig;
@@ -96,6 +97,12 @@
 import org.testng.annotations.Test;
 
 
+class TestReadOnlyStoreSubclass extends ReadOnlyStore {
+  TestReadOnlyStoreSubclass(Store delegate) {
+    super(delegate);
+  }
+}
+
 public class TestMetaDataHandler {
   private static final String ZK_ADDRESS = "localhost:1234";
   private static final String KAFKA_BOOTSTRAP_SERVERS = "localhost:1234";
@@ -1937,4 +1944,51 @@ public void testStoreNamesLookupEmpty() throws IOException {
     MultiStoreResponse storeResponse = OBJECT_MAPPER.readValue(response.content().array(), MultiStoreResponse.class);
     Assert.assertEquals(storeResponse.getStores().length, 0);
   }
+
+  @Test
+  public void testStoreStateLookupWithReadOnlyStore() throws IOException {
+    String storeName = "test-store";
+    Store zkStore = TestUtils.createTestStore(storeName, "test-owner", System.currentTimeMillis());
+    zkStore.setCurrentVersion(1);
+    Store readOnlyStore = new ReadOnlyStore(zkStore);
+
+    HelixReadOnlyStoreRepository storeRepository = Mockito.mock(HelixReadOnlyStoreRepository.class);
+    Mockito.doReturn(readOnlyStore).when(storeRepository).getStore(storeName);
+
+    FullHttpResponse response = passRequestToMetadataHandler(
+        "http://myRouterHost:4567/" + TYPE_STORE_STATE + "/" + storeName,
+        null,
+        null,
+        Mockito.mock(HelixReadOnlyStoreConfigRepository.class),
+        Collections.emptyMap(),
+        Collections.emptyMap(),
+        storeRepository);
+
+    Assert.assertEquals(response.status(), HttpResponseStatus.OK);
+    StoreJSONSerializer serializer = new StoreJSONSerializer();
+    Store deserializedStore = serializer.deserialize(response.content().array(), null);
+    Assert.assertEquals(deserializedStore.getName(), storeName);
+    Assert.assertEquals(deserializedStore.getCurrentVersion(), 1);
+  }
+
+  @Test
+  public void testStoreStateLookupThrowsForReadOnlyStoreSubclass() throws IOException {
+    String storeName = "test-store";
+    Store zkStore = TestUtils.createTestStore(storeName, "test-owner", System.currentTimeMillis());
+    TestReadOnlyStoreSubclass readOnlyStoreSubclass = new TestReadOnlyStoreSubclass(zkStore);
+
+    HelixReadOnlyStoreRepository storeRepository = Mockito.mock(HelixReadOnlyStoreRepository.class);
+    Mockito.doReturn(readOnlyStoreSubclass).when(storeRepository).getStore(storeName);
+
+    Assert.assertThrows(
+        VeniceException.class,
+        () -> passRequestToMetadataHandler(
+            "http://myRouterHost:4567/" + TYPE_STORE_STATE + "/" + storeName,
+            null,
+            null,
+            Mockito.mock(HelixReadOnlyStoreConfigRepository.class),
+            Collections.emptyMap(),
+            Collections.emptyMap(),
+            storeRepository));
+  }
 }