[venice-samza] Enabled pipelineV2 for the d2 client being used in samza producer (#1728)

`pipelineV2` is required in D2 Client to talk H2 to the backend.

In the future, we should construct a D2Client from the internal
factory, where we can manage all the D2 configs in a centralized
fashion.

Author: gaojieliu

integrations/venice-samza/src/main/java/com/linkedin/venice/samza/VeniceSystemProducer.java

@@ -10,6 +10,8 @@
 import com.linkedin.avroutil1.compatibility.AvroCompatibilityHelper;
 import com.linkedin.d2.balancer.D2Client;
 import com.linkedin.d2.balancer.D2ClientBuilder;
+import com.linkedin.r2.transport.common.TransportClientFactory;
+import com.linkedin.r2.transport.http.client.HttpClientFactory;
 import com.linkedin.venice.D2.D2ClientUtils;
 import com.linkedin.venice.client.store.ClientConfig;
 import com.linkedin.venice.client.store.ClientFactory;
@@ -933,6 +935,16 @@ protected void setControllerClient(D2ControllerClient controllerClient) {
   }
 
   private D2Client getStartedD2Client(String d2ZkHost) {
+    /**
+     * Create {@link HttpClientFactory} with {@link com.linkedin.r2.transport.http.client.HttpClientFactory#_usePipelineV2} enabled,
+     * so that it can use the right clients when H2 is enabled.
+     *
+     * TODO: leverage the internal factory to create a proper D2 Client, so that it can follow the default global config.
+     */
+    final Map<String, TransportClientFactory> clientFactories = new HashMap<>();
+    TransportClientFactory transportClientFactory = new HttpClientFactory.Builder().setUsePipelineV2(true).build();
+    clientFactories.put("http", transportClientFactory);
+    clientFactories.put("https", transportClientFactory);
     D2ClientEnvelope d2ClientEnvelope = d2ZkHostToClientEnvelopeMap.computeIfAbsent(d2ZkHost, zkHost -> {
       String fsBasePath = Utils.getUniqueTempPath("d2");
       D2Client d2Client = new D2ClientBuilder().setZkHosts(d2ZkHost)
@@ -941,6 +953,8 @@ private D2Client getStartedD2Client(String d2ZkHost) {
           .setSSLParameters(sslFactory.map(SSLFactory::getSSLParameters).orElse(null))
           .setFsBasePath(fsBasePath)
           .setEnableSaveUriDataOnDisk(true)
+          .setClientFactories(clientFactories)
+          .setRestOverStream(true)
           .build();
       D2ClientUtils.startClient(d2Client);
       return new D2ClientEnvelope(d2Client, fsBasePath);