[controller] Fix store creation and handle schema evolution for push …

…job details and heartbeat system stores
nisargthakkar · Jun 28, 2023 · c4dd5ef · c4dd5ef
1 parent ea885cb
commit c4dd5ef
Show file tree

Hide file tree

Showing 45 changed files with 852 additions and 830 deletions.
diff --git a/...e-common/src/main/java/com/linkedin/venice/serialization/avro/AvroProtocolDefinition.java b/...e-common/src/main/java/com/linkedin/venice/serialization/avro/AvroProtocolDefinition.java
@@ -15,6 +15,7 @@
 import com.linkedin.venice.kafka.protocol.state.StoreVersionState;
 import com.linkedin.venice.meta.Store;
 import com.linkedin.venice.metadata.response.MetadataResponseRecord;
+import com.linkedin.venice.participant.protocol.ParticipantMessageValue;
 import com.linkedin.venice.pubsub.api.PubSubPositionWireFormat;
 import com.linkedin.venice.pushstatus.PushStatusKey;
 import com.linkedin.venice.pushstatus.PushStatusValue;
@@ -147,6 +148,11 @@ public enum AvroProtocolDefinition {
    */
   PUSH_STATUS_SYSTEM_SCHEMA_STORE(1, PushStatusValue.class),
 
+  /**
+   * Value schema for participant system stores.
+   */
+  PARTICIPANT_MESSAGE_SYSTEM_STORE_VALUE(1, ParticipantMessageValue.class),
+
   /**
    * Response record for admin request v1.
    */
@@ -230,7 +236,7 @@ private static Set<Byte> validateMagicBytes() {
    * extra header prepended in front. These are either un-evolvable or have their
    * schema ID stored out of band from the record.
    *
-   * For example, everything that goes inside of the Put of a {@link KafkaMessageEnvelope}
+   * For example, everything that goes inside the Put message of a {@link KafkaMessageEnvelope}
    * and uses the {@link com.linkedin.venice.kafka.protocol.Put#schemaId} to store
    * the protocol version.
    */

diff --git a/...rationTest/java/com/linkedin/venice/consumer/ConsumerIntegrationTestWithSchemaReader.java b/...rationTest/java/com/linkedin/venice/consumer/ConsumerIntegrationTestWithSchemaReader.java
@@ -1,6 +1,5 @@
 package com.linkedin.venice.consumer;
 
-import com.linkedin.venice.controller.VeniceHelixAdmin;
 import com.linkedin.venice.controllerapi.ControllerClient;
 import com.linkedin.venice.controllerapi.MultiSchemaResponse;
 import com.linkedin.venice.integration.utils.VeniceClusterWrapper;
@@ -32,13 +31,15 @@ void extraBeforeClassSetUp(VeniceClusterWrapper cluster, ControllerClient contro
           AvroProtocolDefinition.KAFKA_MESSAGE_ENVELOPE.getCurrentProtocolVersion());
     });
 
-    ((VeniceHelixAdmin) cluster.getRandomVeniceController().getVeniceAdmin()).addValueSchema(
-        cluster.getClusterName(),
-        systemStoreName,
-        NEW_PROTOCOL_SCHEMA.toString(),
-        NEW_PROTOCOL_VERSION,
-        DirectionalSchemaCompatibilityType.NONE,
-        false);
+    cluster.getRandomVeniceController()
+        .getVeniceAdmin()
+        .addValueSchemaInternal(
+            cluster.getClusterName(),
+            systemStoreName,
+            NEW_PROTOCOL_SCHEMA.toString(),
+            NEW_PROTOCOL_VERSION,
+            DirectionalSchemaCompatibilityType.NONE,
+            false);
   }
 
   @Override

diff --git a/...common/src/integrationTest/java/com/linkedin/venice/controller/AdminToolBackfillTest.java b/...common/src/integrationTest/java/com/linkedin/venice/controller/AdminToolBackfillTest.java
@@ -17,7 +17,6 @@
 import com.linkedin.venice.integration.utils.VeniceTwoLayerMultiRegionMultiClusterWrapper;
 import com.linkedin.venice.utils.TestUtils;
 import com.linkedin.venice.utils.Utils;
-import com.linkedin.venice.utils.VeniceProperties;
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
@@ -56,7 +55,7 @@ public void setUp() {
         2,
         2,
         1,
-        Optional.of(new VeniceProperties(parentControllerProperties)),
+        Optional.of(parentControllerProperties),
         Optional.empty(),
         Optional.empty(),
         false);

diff --git a/...st-common/src/integrationTest/java/com/linkedin/venice/controller/TestFabricBuildout.java b/...st-common/src/integrationTest/java/com/linkedin/venice/controller/TestFabricBuildout.java
@@ -22,7 +22,6 @@
 import com.linkedin.venice.utils.TestUtils;
 import com.linkedin.venice.utils.TestWriteUtils;
 import com.linkedin.venice.utils.Utils;
-import com.linkedin.venice.utils.VeniceProperties;
 import java.util.List;
 import java.util.Optional;
 import java.util.Properties;
@@ -63,7 +62,7 @@ public void setUp() {
         1,
         Optional.empty(),
         Optional.of(childControllerProperties),
-        Optional.of(new VeniceProperties(serverProperties)),
+        Optional.of(serverProperties),
         false);
 
     childDatacenters = multiRegionMultiClusterWrapper.getChildRegions();

diff --git a/...tionTest/java/com/linkedin/venice/controller/TestParentControllerWithMultiDataCenter.java b/...tionTest/java/com/linkedin/venice/controller/TestParentControllerWithMultiDataCenter.java
@@ -24,7 +24,6 @@
 import com.linkedin.venice.utils.TestUtils;
 import com.linkedin.venice.utils.TestWriteUtils;
 import com.linkedin.venice.utils.Utils;
-import com.linkedin.venice.utils.VeniceProperties;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -71,7 +70,7 @@ public void setUp() {
         1,
         1,
         1,
-        Optional.of(new VeniceProperties(controllerProps)),
+        Optional.of(controllerProps),
         Optional.of(controllerProps),
         Optional.empty());
 

diff --git a/...n/src/integrationTest/java/com/linkedin/venice/controller/VeniceParentHelixAdminTest.java b/...n/src/integrationTest/java/com/linkedin/venice/controller/VeniceParentHelixAdminTest.java
diff --git a/...tegrationTest/java/com/linkedin/venice/endToEnd/ActiveActiveReplicationForHybridTest.java b/...tegrationTest/java/com/linkedin/venice/endToEnd/ActiveActiveReplicationForHybridTest.java
@@ -154,9 +154,9 @@ public void setUp() {
         2,
         1,
         2,
-        Optional.of(new VeniceProperties(controllerProps)),
         Optional.of(controllerProps),
-        Optional.of(new VeniceProperties(serverProperties)),
+        Optional.of(controllerProps),
+        Optional.of(serverProperties),
         false);
     childDatacenters = multiRegionMultiClusterWrapper.getChildRegions();
     parentControllers = multiRegionMultiClusterWrapper.getParentControllers();

diff --git a/...mon/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciClusterAgnosticTest.java b/...mon/src/integrationTest/java/com/linkedin/venice/endToEnd/DaVinciClusterAgnosticTest.java
@@ -35,9 +35,9 @@
 import io.tehuti.metrics.MetricsRepository;
 import java.util.AbstractMap;
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
+import java.util.Properties;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -72,6 +72,8 @@ public class DaVinciClusterAgnosticTest {
   @BeforeClass
   public void setUp() {
     Utils.thisIsLocalhost();
+    Properties parentControllerProps = new Properties();
+    parentControllerProps.put(OFFLINE_JOB_START_TIMEOUT_MS, "180000");
     multiRegionMultiClusterWrapper = ServiceFactory.getVeniceTwoLayerMultiRegionMultiClusterWrapper(
         1,
         2,
@@ -80,7 +82,7 @@ public void setUp() {
         3,
         1,
         3,
-        Optional.of(new VeniceProperties(Collections.singletonMap(OFFLINE_JOB_START_TIMEOUT_MS, "180000"))),
+        Optional.of(parentControllerProps),
         Optional.empty(),
         Optional.empty(),
         false);

diff --git a/...e-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DataRecoveryTest.java b/...e-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/DataRecoveryTest.java
@@ -48,7 +48,6 @@
 import com.linkedin.venice.samza.VeniceSystemFactory;
 import com.linkedin.venice.utils.TestUtils;
 import com.linkedin.venice.utils.Utils;
-import com.linkedin.venice.utils.VeniceProperties;
 import java.util.AbstractMap;
 import java.util.Arrays;
 import java.util.HashMap;
@@ -107,9 +106,9 @@ public void setUp() {
         2,
         1,
         2,
-        Optional.of(new VeniceProperties(controllerProps)),
         Optional.of(controllerProps),
-        Optional.of(new VeniceProperties(serverProperties)),
+        Optional.of(controllerProps),
+        Optional.of(serverProperties),
         false);
     childDatacenters = multiRegionMultiClusterWrapper.getChildRegions();
     parentControllers = multiRegionMultiClusterWrapper.getParentControllers();

diff --git a/...-common/src/integrationTest/java/com/linkedin/venice/endToEnd/NearlineE2ELatencyTest.java b/...-common/src/integrationTest/java/com/linkedin/venice/endToEnd/NearlineE2ELatencyTest.java
@@ -1,8 +1,9 @@
 package com.linkedin.venice.endToEnd;
 
-import static com.linkedin.venice.ConfigKeys.*;
-import static com.linkedin.venice.utils.IntegrationTestPushUtils.*;
-import static com.linkedin.venice.utils.TestWriteUtils.*;
+import static com.linkedin.venice.ConfigKeys.SERVER_PROMOTION_TO_LEADER_REPLICA_DELAY_SECONDS;
+import static com.linkedin.venice.utils.IntegrationTestPushUtils.getSamzaProducer;
+import static com.linkedin.venice.utils.IntegrationTestPushUtils.sendStreamingRecord;
+import static com.linkedin.venice.utils.TestWriteUtils.STRING_SCHEMA;
 
 import com.linkedin.davinci.stats.IngestionStats;
 import com.linkedin.venice.client.store.AvroGenericStoreClient;
@@ -23,7 +24,6 @@
 import com.linkedin.venice.server.VeniceServer;
 import com.linkedin.venice.utils.TestUtils;
 import com.linkedin.venice.utils.Utils;
-import com.linkedin.venice.utils.VeniceProperties;
 import io.tehuti.metrics.MetricsRepository;
 import java.util.AbstractMap;
 import java.util.Arrays;
@@ -77,7 +77,7 @@ public void setUp() {
         REPLICATION_FACTOR,
         Optional.empty(),
         Optional.empty(),
-        Optional.of(new VeniceProperties(serverProperties)),
+        Optional.of(serverProperties),
         false);
     childDatacenters = multiRegionMultiClusterWrapper.getChildRegions();
     parentControllers = multiRegionMultiClusterWrapper.getParentControllers();

diff --git a/...ommon/src/integrationTest/java/com/linkedin/venice/endToEnd/OneTouchDataRecoveryTest.java b/...ommon/src/integrationTest/java/com/linkedin/venice/endToEnd/OneTouchDataRecoveryTest.java
@@ -17,7 +17,6 @@
 import com.linkedin.venice.meta.Version;
 import com.linkedin.venice.utils.TestUtils;
 import com.linkedin.venice.utils.Utils;
-import com.linkedin.venice.utils.VeniceProperties;
 import java.util.AbstractMap;
 import java.util.Arrays;
 import java.util.List;
@@ -67,7 +66,7 @@ public void setUp() {
         REPLICATION_FACTOR,
         Optional.empty(),
         Optional.empty(),
-        Optional.of(new VeniceProperties(serverProperties)),
+        Optional.of(serverProperties),
         false);
 
     childDatacenters = multiRegionMultiClusterWrapper.getChildRegions();

diff --git a/...-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/PartialUpdateTest.java b/...-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/PartialUpdateTest.java
@@ -72,7 +72,6 @@
 import com.linkedin.venice.utils.TestWriteUtils;
 import com.linkedin.venice.utils.Time;
 import com.linkedin.venice.utils.Utils;
-import com.linkedin.venice.utils.VeniceProperties;
 import com.linkedin.venice.writer.update.UpdateBuilder;
 import com.linkedin.venice.writer.update.UpdateBuilderImpl;
 import java.io.ByteArrayOutputStream;
@@ -130,9 +129,9 @@ public void setUp() {
         2,
         1,
         2,
-        Optional.of(new VeniceProperties(controllerProps)),
+        Optional.of(controllerProps),
         Optional.of(new Properties(controllerProps)),
-        Optional.of(new VeniceProperties(serverProperties)),
+        Optional.of(serverProperties),
         false);
     this.childDatacenters = multiRegionMultiClusterWrapper.getChildRegions();
     List<VeniceControllerWrapper> parentControllers = multiRegionMultiClusterWrapper.getParentControllers();

diff --git a/...test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/PushJobDetailsTest.java b/...test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/PushJobDetailsTest.java
@@ -50,7 +50,6 @@
 import com.linkedin.venice.utils.TestUtils;
 import com.linkedin.venice.utils.Time;
 import com.linkedin.venice.utils.Utils;
-import com.linkedin.venice.utils.VeniceProperties;
 import java.io.File;
 import java.io.IOException;
 import java.util.Arrays;
@@ -96,9 +95,9 @@ public void setUp() throws IOException {
         1,
         1,
         1,
-        Optional.of(new VeniceProperties(parentControllerProperties)),
+        Optional.of(parentControllerProperties),
         Optional.empty(),
-        Optional.of(new VeniceProperties(serverProperties)),
+        Optional.of(serverProperties),
         false);
     String clusterName = multiRegionMultiClusterWrapper.getClusterNames()[0];
 

diff --git a/...n/src/integrationTest/java/com/linkedin/venice/endToEnd/PushStatusStoreMultiColoTest.java b/...n/src/integrationTest/java/com/linkedin/venice/endToEnd/PushStatusStoreMultiColoTest.java
@@ -33,7 +33,6 @@
 import com.linkedin.venice.serialization.avro.AvroProtocolDefinition;
 import com.linkedin.venice.utils.TestUtils;
 import com.linkedin.venice.utils.Utils;
-import com.linkedin.venice.utils.VeniceProperties;
 import java.util.List;
 import java.util.Optional;
 import java.util.Properties;
@@ -85,7 +84,7 @@ public void setUp() {
         NUMBER_OF_SERVERS,
         NUMBER_OF_ROUTERS,
         REPLICATION_FACTOR,
-        Optional.of(new VeniceProperties(extraProperties)),
+        Optional.of(extraProperties),
         Optional.of(extraProperties),
         Optional.empty(),
         false);

diff --git a/...mmon/src/integrationTest/java/com/linkedin/venice/endToEnd/TestActiveActiveIngestion.java b/...mmon/src/integrationTest/java/com/linkedin/venice/endToEnd/TestActiveActiveIngestion.java
@@ -70,7 +70,6 @@
 import com.linkedin.venice.utils.TestWriteUtils;
 import com.linkedin.venice.utils.Time;
 import com.linkedin.venice.utils.Utils;
-import com.linkedin.venice.utils.VeniceProperties;
 import com.linkedin.venice.view.TestView;
 import com.linkedin.venice.views.ChangeCaptureView;
 import com.linkedin.venice.writer.VeniceWriter;
@@ -137,7 +136,7 @@ public void setUp() {
         1,
         Optional.empty(),
         Optional.empty(),
-        Optional.of(new VeniceProperties(serverProperties)),
+        Optional.of(serverProperties),
         false);
 
     childDatacenters = multiRegionMultiClusterWrapper.getChildRegions();

diff --git a/...egrationTest/java/com/linkedin/venice/endToEnd/TestActiveActiveReplicationForIncPush.java b/...egrationTest/java/com/linkedin/venice/endToEnd/TestActiveActiveReplicationForIncPush.java
@@ -26,7 +26,6 @@
 import com.linkedin.venice.utils.TestWriteUtils;
 import com.linkedin.venice.utils.Time;
 import com.linkedin.venice.utils.Utils;
-import com.linkedin.venice.utils.VeniceProperties;
 import java.io.File;
 import java.util.List;
 import java.util.Optional;
@@ -83,9 +82,9 @@ public void setUp() {
         2,
         1,
         2,
-        Optional.of(new VeniceProperties(controllerProps)),
         Optional.of(controllerProps),
-        Optional.of(new VeniceProperties(serverProperties)),
+        Optional.of(controllerProps),
+        Optional.of(serverProperties),
         false);
     childDatacenters = multiRegionMultiClusterWrapper.getChildRegions();
     clusterNames = multiRegionMultiClusterWrapper.getClusterNames();

diff --git a/...tionTest/java/com/linkedin/venice/endToEnd/TestActiveActiveReplicationWithDownRegion.java b/...tionTest/java/com/linkedin/venice/endToEnd/TestActiveActiveReplicationWithDownRegion.java
@@ -29,7 +29,6 @@
 import com.linkedin.venice.samza.VeniceSystemProducer;
 import com.linkedin.venice.utils.TestUtils;
 import com.linkedin.venice.utils.Utils;
-import com.linkedin.venice.utils.VeniceProperties;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -100,9 +99,9 @@ public void setUp() {
         2,
         1,
         2,
-        Optional.of(new VeniceProperties(controllerProps)),
         Optional.of(controllerProps),
-        Optional.of(new VeniceProperties(serverProperties)),
+        Optional.of(controllerProps),
+        Optional.of(serverProperties),
         false);
     childDatacenters = multiRegionMultiClusterWrapper.getChildRegions();
     parentControllers = multiRegionMultiClusterWrapper.getParentControllers();

diff --git a/...mon/src/integrationTest/java/com/linkedin/venice/endToEnd/TestChangeCaptureIngestion.java b/...mon/src/integrationTest/java/com/linkedin/venice/endToEnd/TestChangeCaptureIngestion.java
@@ -1,14 +1,33 @@
 package com.linkedin.venice.endToEnd;
 
-import static com.linkedin.davinci.store.rocksdb.RocksDBServerConfig.*;
-import static com.linkedin.venice.CommonConfigKeys.*;
-import static com.linkedin.venice.ConfigKeys.*;
-import static com.linkedin.venice.hadoop.VenicePushJob.*;
+import static com.linkedin.davinci.store.rocksdb.RocksDBServerConfig.ROCKSDB_PLAIN_TABLE_FORMAT_ENABLED;
+import static com.linkedin.venice.CommonConfigKeys.SSL_ENABLED;
+import static com.linkedin.venice.ConfigKeys.CHILD_DATA_CENTER_KAFKA_URL_PREFIX;
+import static com.linkedin.venice.hadoop.VenicePushJob.DEFAULT_KEY_FIELD_PROP;
+import static com.linkedin.venice.hadoop.VenicePushJob.DEFAULT_VALUE_FIELD_PROP;
+import static com.linkedin.venice.hadoop.VenicePushJob.KAFKA_INPUT_BROKER_URL;
+import static com.linkedin.venice.hadoop.VenicePushJob.KAFKA_INPUT_MAX_RECORDS_PER_MAPPER;
+import static com.linkedin.venice.hadoop.VenicePushJob.REPUSH_TTL_ENABLE;
+import static com.linkedin.venice.hadoop.VenicePushJob.REWIND_TIME_IN_SECONDS_OVERRIDE;
+import static com.linkedin.venice.hadoop.VenicePushJob.SOURCE_KAFKA;
 import static com.linkedin.venice.integration.utils.VeniceClusterWrapperConstants.STANDALONE_REGION_NAME;
-import static com.linkedin.venice.integration.utils.VeniceControllerWrapper.*;
-import static com.linkedin.venice.samza.VeniceSystemFactory.*;
-import static com.linkedin.venice.utils.IntegrationTestPushUtils.*;
-import static com.linkedin.venice.utils.TestWriteUtils.*;
+import static com.linkedin.venice.integration.utils.VeniceControllerWrapper.D2_SERVICE_NAME;
+import static com.linkedin.venice.integration.utils.VeniceControllerWrapper.PARENT_D2_SERVICE_NAME;
+import static com.linkedin.venice.samza.VeniceSystemFactory.DEPLOYMENT_ID;
+import static com.linkedin.venice.samza.VeniceSystemFactory.DOT;
+import static com.linkedin.venice.samza.VeniceSystemFactory.SYSTEMS_PREFIX;
+import static com.linkedin.venice.samza.VeniceSystemFactory.VENICE_AGGREGATE;
+import static com.linkedin.venice.samza.VeniceSystemFactory.VENICE_CHILD_CONTROLLER_D2_SERVICE;
+import static com.linkedin.venice.samza.VeniceSystemFactory.VENICE_CHILD_D2_ZK_HOSTS;
+import static com.linkedin.venice.samza.VeniceSystemFactory.VENICE_PARENT_CONTROLLER_D2_SERVICE;
+import static com.linkedin.venice.samza.VeniceSystemFactory.VENICE_PARENT_D2_ZK_HOSTS;
+import static com.linkedin.venice.samza.VeniceSystemFactory.VENICE_PUSH_TYPE;
+import static com.linkedin.venice.samza.VeniceSystemFactory.VENICE_STORE;
+import static com.linkedin.venice.utils.IntegrationTestPushUtils.createStoreForJob;
+import static com.linkedin.venice.utils.IntegrationTestPushUtils.sendStreamingDeleteRecord;
+import static com.linkedin.venice.utils.IntegrationTestPushUtils.sendStreamingRecord;
+import static com.linkedin.venice.utils.TestWriteUtils.getTempDataDirectory;
+import static com.linkedin.venice.utils.TestWriteUtils.writeSimpleAvroFileWithUserSchema;
 
 import com.linkedin.davinci.consumer.ChangeEvent;
 import com.linkedin.davinci.consumer.ChangelogClientConfig;
@@ -46,7 +65,6 @@
 import com.linkedin.venice.utils.TestWriteUtils;
 import com.linkedin.venice.utils.Time;
 import com.linkedin.venice.utils.Utils;
-import com.linkedin.venice.utils.VeniceProperties;
 import com.linkedin.venice.view.TestView;
 import com.linkedin.venice.views.ChangeCaptureView;
 import com.linkedin.venice.writer.VeniceWriter;
@@ -108,7 +126,7 @@ public void setUp() {
         1,
         Optional.empty(),
         Optional.empty(),
-        Optional.of(new VeniceProperties(serverProperties)),
+        Optional.of(serverProperties),
         false);
 
     childDatacenters = multiRegionMultiClusterWrapper.getChildRegions();