Move server D2 announcement from router to server

The router was announcing the server's D2 URI on behalf of the server,
which is architecturally wrong. Each component should announce itself.

- Add D2 self-announcement to VeniceServer.run() gated by
  server.d2.announce.enabled
- Remove server D2 announcement hack from RouterServer.main()
- Move server D2 config from router.properties to server.properties
- Add cluster.to.server.d2 to controller.properties for discovery response

Author: sushantmane

docker/venice-controller/single-dc-configs/controller.properties

@@ -10,6 +10,7 @@ ssl.to.kakfa=false
 controller.parent.mode=false
 controller.system.schema.cluster.name=venice-cluster0
 cluster.to.d2=venice-cluster0:venice-discovery
+cluster.to.server.d2=venice-cluster0:venice-server-d2
 delay.to.rebalance.ms=0
 offline.job.start.timeout.ms=60000
 topic.cleanup.sleep.interval.between.topic.list.fetch.ms=30000

docker/venice-router/single-dc-configs/router.properties

@@ -15,8 +15,6 @@ router.max.outgoing.connection.per.route=2
 router.http.client.pool.size=2
 cluster.to.d2=venice-cluster0:venice-discovery
 cluster.to.server.d2=venice-cluster0:venice-server-d2
-server.d2.announce.host=venice-server
-server.d2.announce.port=54093
 kafka.bootstrap.servers=kafka:9092
 router.storage.node.client.type=APACHE_HTTP_ASYNC_CLIENT
 router.io.worker.count=4

docker/venice-server/single-dc-configs/server.properties

@@ -14,6 +14,9 @@ kafka.read.cycle.delay.ms=50
 disk.full.threshold=0.99
 server.partition.graceful.drop.time.in.seconds=0
 system.schema.cluster.name=venice-cluster0
+server.d2.announce.enabled=true
+server.d2.announce.host=venice-server
+server.d2.service.name=venice-server-d2
 data.base.path=/opt/venice/rocksdb
 server.ingestion.isolation.application.port=54094
 persistence.type=ROCKS_DB

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

@@ -274,25 +274,6 @@ public static void run(String routerConfigFilePath, boolean runForever) throws E
         d2Servers.addAll(D2ConfigUtils.getD2Servers(zkAddress, discoveryClusterName, localUri));
       }
 
-      Map<String, String> clusterToServerD2 = props.getMap(ConfigKeys.CLUSTER_TO_SERVER_D2, Collections.emptyMap());
-      String serverAnnounceHost = props.getString("server.d2.announce.host", "");
-      int serverAnnouncePort = props.getInt("server.d2.announce.port", 0);
-      for (Map.Entry<String, String> entry: clusterToServerD2.entrySet()) {
-        String d2ServiceName = entry.getValue();
-        String d2ClusterName = d2ServiceName + "_d2_cluster";
-        D2ConfigUtils.setupD2Config(zkAddress, false, d2ClusterName, d2ServiceName);
-        if (!serverAnnounceHost.isEmpty() && serverAnnouncePort > 0) {
-          String serverUri = "http://" + serverAnnounceHost + ":" + serverAnnouncePort;
-          d2Servers.addAll(D2ConfigUtils.getD2Servers(zkAddress, d2ClusterName, serverUri));
-          LOGGER.info("Server D2 announcement for service {} at URI: {}", d2ServiceName, serverUri);
-        } else {
-          LOGGER.warn(
-              "server.d2.announce.host/port not set; skipping D2 URI announcement for server service {}. "
-                  + "D2 config will be written to ZK but no URI will be announced from the router.",
-              d2ServiceName);
-        }
-      }
-
       LOGGER.info("D2 announcement enabled with {} announcers for router URI: {}", d2Servers.size(), localUri);
     }
 

services/venice-server/src/main/java/com/linkedin/venice/server/VeniceServer.java

@@ -30,6 +30,7 @@
 import com.linkedin.davinci.storage.StorageEngineRepository;
 import com.linkedin.davinci.storage.StorageMetadataService;
 import com.linkedin.davinci.storage.StorageService;
+import com.linkedin.venice.ConfigKeys;
 import com.linkedin.venice.acl.DynamicAccessController;
 import com.linkedin.venice.acl.StaticAccessController;
 import com.linkedin.venice.cleaner.BackupVersionOptimizationService;
@@ -38,6 +39,7 @@
 import com.linkedin.venice.client.store.ClientConfig;
 import com.linkedin.venice.client.store.ClientFactory;
 import com.linkedin.venice.common.VeniceSystemStoreUtils;
+import com.linkedin.venice.d2.D2ConfigUtils;
 import com.linkedin.venice.exceptions.VeniceException;
 import com.linkedin.venice.helix.AllowlistAccessor;
 import com.linkedin.venice.helix.HelixCustomizedViewOfflinePushRepository;
@@ -71,6 +73,7 @@
 import com.linkedin.venice.system.store.ControllerClientBackedSystemSchemaInitializer;
 import com.linkedin.venice.utils.CollectionUtils;
 import com.linkedin.venice.utils.Utils;
+import com.linkedin.venice.utils.VeniceProperties;
 import com.linkedin.venice.utils.lazy.Lazy;
 import io.tehuti.metrics.MetricsRepository;
 import java.util.ArrayList;
@@ -854,8 +857,25 @@ public static void run(String configDirectory, boolean joinThread) throws Except
   }
 
   public static void run(VeniceConfigLoader veniceConfigService, boolean joinThread) throws Exception {
-    VeniceServerContext serverContext =
-        new VeniceServerContext.Builder().setVeniceConfigLoader(veniceConfigService).build();
+    List<ServiceDiscoveryAnnouncer> d2Servers = new ArrayList<>();
+    VeniceProperties props = veniceConfigService.getCombinedProperties();
+
+    if (props.getBoolean("server.d2.announce.enabled", false)) {
+      String zkAddress = props.getString(ConfigKeys.ZOOKEEPER_ADDRESS);
+      int port = props.getInt(ConfigKeys.LISTENER_PORT);
+      String announceHost = props.getString("server.d2.announce.host", "localhost");
+      String localUri = "http://" + announceHost + ":" + port;
+
+      String d2ServiceName = props.getString("server.d2.service.name", "venice-server-d2");
+      String d2ClusterName = d2ServiceName + "_d2_cluster";
+      D2ConfigUtils.setupD2Config(zkAddress, false, d2ClusterName, d2ServiceName);
+      d2Servers.addAll(D2ConfigUtils.getD2Servers(zkAddress, d2ClusterName, localUri));
+      LOGGER.info("Server D2 announcement enabled for service {} at URI: {}", d2ServiceName, localUri);
+    }
+
+    VeniceServerContext serverContext = new VeniceServerContext.Builder().setVeniceConfigLoader(veniceConfigService)
+        .setServiceDiscoveryAnnouncers(d2Servers)
+        .build();
     final VeniceServer server = new VeniceServer(serverContext);
     if (!server.isStarted()) {
       server.start();