-
Notifications
You must be signed in to change notification settings - Fork 91
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
add requestbasedmetarepository integration testing
- Loading branch information
Showing
14 changed files
with
538 additions
and
107 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
264 changes: 264 additions & 0 deletions
264
...egrationTest/java/com/linkedin/venice/endToEnd/TestDaVinciRequestBasedMetaRepository.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,264 @@ | ||
| package com.linkedin.venice.endToEnd; | ||
|
|
||
| import static com.linkedin.venice.ConfigKeys.CLIENT_SYSTEM_STORE_REPOSITORY_REFRESH_INTERVAL_SECONDS; | ||
| import static com.linkedin.venice.ConfigKeys.CLIENT_USE_SYSTEM_STORE_REPOSITORY; | ||
| import static com.linkedin.venice.ConfigKeys.DATA_BASE_PATH; | ||
| import static com.linkedin.venice.ConfigKeys.PERSISTENCE_TYPE; | ||
| import static com.linkedin.venice.utils.TestWriteUtils.getTempDataDirectory; | ||
| import static com.linkedin.venice.vpj.VenicePushJobConstants.DEFAULT_VALUE_FIELD_PROP; | ||
| import static org.testng.Assert.assertNotNull; | ||
|
|
||
| import com.linkedin.d2.balancer.D2Client; | ||
| import com.linkedin.davinci.client.DaVinciClient; | ||
| import com.linkedin.davinci.client.DaVinciConfig; | ||
| import com.linkedin.davinci.client.factory.CachingDaVinciClientFactory; | ||
| import com.linkedin.venice.D2.D2ClientUtils; | ||
| import com.linkedin.venice.controllerapi.ControllerClient; | ||
| import com.linkedin.venice.controllerapi.SchemaResponse; | ||
| import com.linkedin.venice.integration.utils.D2TestUtils; | ||
| import com.linkedin.venice.integration.utils.ServiceFactory; | ||
| import com.linkedin.venice.integration.utils.VeniceClusterCreateOptions; | ||
| import com.linkedin.venice.integration.utils.VeniceClusterWrapper; | ||
| import com.linkedin.venice.integration.utils.VeniceRouterWrapper; | ||
| import com.linkedin.venice.meta.PersistenceType; | ||
| import com.linkedin.venice.utils.IntegrationTestPushUtils; | ||
| import com.linkedin.venice.utils.PropertyBuilder; | ||
| import com.linkedin.venice.utils.TestUtils; | ||
| 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 io.tehuti.Metric; | ||
| import io.tehuti.metrics.MetricsRepository; | ||
| import java.io.File; | ||
| import java.io.IOException; | ||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
| import java.util.Properties; | ||
| import java.util.concurrent.ExecutionException; | ||
| import java.util.concurrent.TimeUnit; | ||
| import org.apache.avro.Schema; | ||
| import org.testng.Assert; | ||
| import org.testng.annotations.AfterClass; | ||
| import org.testng.annotations.BeforeClass; | ||
| import org.testng.annotations.Test; | ||
|
|
||
|
|
||
| public class TestDaVinciRequestBasedMetaRepository { | ||
| private static final int TEST_TIMEOUT = 2 * Time.MS_PER_MINUTE; | ||
|
|
||
| private static final String CLUSTER_NAME = "venice-cluster"; | ||
| private VeniceClusterWrapper clusterWrapper; | ||
|
|
||
| private static final String storeNameStringToString = "store-name-string-to-string"; | ||
| private static final String storeNameStringToNameRecord = "store-name-string-to-name-record"; | ||
|
|
||
| // StoreName -> ControllerClient | ||
| // Using map to check which stores are created | ||
| private final Map<String, ControllerClient> controllerClients = new HashMap<>(); | ||
| // StoreName -> Directory | ||
| private final Map<String, File> pushJobAvroDataDirs = new HashMap<>(); | ||
|
|
||
| private DaVinciConfig daVinciConfig; | ||
| private MetricsRepository dvcMetricsRepo; | ||
| private D2Client daVinciD2RemoteFabric; | ||
| private CachingDaVinciClientFactory daVinciClientFactory; | ||
|
|
||
| @BeforeClass(alwaysRun = true) | ||
| public void setUp() throws IOException { | ||
|
|
||
| VeniceClusterCreateOptions.Builder options = new VeniceClusterCreateOptions.Builder().clusterName(CLUSTER_NAME) | ||
| .numberOfRouters(1) | ||
| .numberOfServers(2) | ||
| .numberOfControllers(2) | ||
| .replicationFactor(2) | ||
| .forkServer(false); | ||
| clusterWrapper = ServiceFactory.getVeniceCluster(options.build()); | ||
|
|
||
| // Create stores | ||
| runPushJob( // String to String | ||
| storeNameStringToString, | ||
| TestWriteUtils | ||
| .writeSimpleAvroFileWithStringToStringSchema(getPushJobAvroFileDirectory(storeNameStringToString))); | ||
| runPushJob( // String to Name Record | ||
| storeNameStringToNameRecord, | ||
| TestWriteUtils.writeSimpleAvroFileWithStringToNameRecordV1Schema( | ||
| getPushJobAvroFileDirectory(storeNameStringToNameRecord))); | ||
|
|
||
| // Set up DVC Client Factory | ||
| VeniceProperties backendConfig = | ||
| new PropertyBuilder().put(DATA_BASE_PATH, Utils.getTempDataDirectory().getAbsolutePath()) | ||
| .put(PERSISTENCE_TYPE, PersistenceType.ROCKS_DB) | ||
| .put(CLIENT_USE_SYSTEM_STORE_REPOSITORY, true) | ||
| .put(CLIENT_SYSTEM_STORE_REPOSITORY_REFRESH_INTERVAL_SECONDS, 1) | ||
| .build(); | ||
| daVinciConfig = new DaVinciConfig(); | ||
| daVinciConfig.setUseRequestBasedMetaRepository(true); | ||
| daVinciD2RemoteFabric = D2TestUtils.getAndStartD2Client(clusterWrapper.getZk().getAddress()); | ||
| dvcMetricsRepo = new MetricsRepository(); | ||
| daVinciClientFactory = new CachingDaVinciClientFactory( | ||
| daVinciD2RemoteFabric, | ||
| VeniceRouterWrapper.CLUSTER_DISCOVERY_D2_SERVICE_NAME, | ||
| dvcMetricsRepo, | ||
| backendConfig); | ||
| } | ||
|
|
||
| @AfterClass(alwaysRun = true) | ||
| public void cleanUp() { | ||
|
|
||
| // Shutdown remote fabric | ||
| D2ClientUtils.shutdownClient(daVinciD2RemoteFabric); | ||
|
|
||
| // Close client factory | ||
| daVinciClientFactory.close(); | ||
|
|
||
| // Close controller clients | ||
| for (Map.Entry<String, ControllerClient> entry: controllerClients.entrySet()) { | ||
| entry.getValue().close(); | ||
| } | ||
|
|
||
| // Close cluster wrapper | ||
| clusterWrapper.close(); | ||
| } | ||
|
|
||
| @Test(timeOut = TEST_TIMEOUT) | ||
| public void testDVCRequestBasedMetaRepositoryStringToString() | ||
| throws IOException, ExecutionException, InterruptedException { | ||
|
|
||
| try (DaVinciClient<String, Object> storeClient = | ||
| daVinciClientFactory.getAndStartGenericAvroClient(storeNameStringToString, daVinciConfig)) { | ||
| storeClient.subscribeAll().get(); | ||
|
|
||
| int recordCount = TestWriteUtils.DEFAULT_USER_DATA_RECORD_COUNT; | ||
| for (int i = 1; i <= recordCount; i++) { | ||
| Assert.assertEquals( | ||
| storeClient.get(Integer.toString(i)).get().toString(), | ||
| TestWriteUtils.DEFAULT_USER_DATA_VALUE_PREFIX + i); | ||
| } | ||
| Assert | ||
| .assertEquals(getMetric(dvcMetricsRepo, "current_version_number.Gauge", storeNameStringToString), (double) 1); | ||
|
|
||
| // Perform another push with 200 keys to verify future version ingestion | ||
| recordCount = 200; | ||
| runPushJob( | ||
| storeNameStringToString, | ||
| TestWriteUtils.writeSimpleAvroFileWithStringToStringSchema( | ||
| getPushJobAvroFileDirectory(storeNameStringToString), | ||
| recordCount)); | ||
|
|
||
| // Perform another push with 200 keys to verify future version ingestion and version swap | ||
| TestUtils.waitForNonDeterministicAssertion(60, TimeUnit.SECONDS, false, () -> { | ||
| Assert.assertEquals( | ||
| getMetric(dvcMetricsRepo, "current_version_number.Gauge", storeNameStringToString), | ||
| (double) 2); | ||
| }); | ||
|
|
||
| for (int i = 1; i <= recordCount; i++) { | ||
| Assert.assertEquals( | ||
| storeClient.get(Integer.toString(i)).get().toString(), | ||
| TestWriteUtils.DEFAULT_USER_DATA_VALUE_PREFIX + i); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| @Test(timeOut = TEST_TIMEOUT) | ||
| public void testDVCRequestBasedMetaRepositoryStringToNameRecord() throws ExecutionException, InterruptedException { | ||
|
|
||
| try (DaVinciClient<String, Object> storeClient = | ||
| daVinciClientFactory.getAndStartGenericAvroClient(storeNameStringToNameRecord, daVinciConfig)) { | ||
| storeClient.subscribeAll().get(); | ||
|
|
||
| int recordCount = TestWriteUtils.DEFAULT_USER_DATA_RECORD_COUNT; | ||
| for (int i = 1; i <= recordCount; i++) { | ||
| Assert.assertEquals( | ||
| storeClient.get(Integer.toString(i)).get().toString(), | ||
| TestWriteUtils.renderNameRecord(TestWriteUtils.STRING_TO_NAME_RECORD_V1_SCHEMA, i) | ||
| .get(DEFAULT_VALUE_FIELD_PROP) | ||
| .toString()); | ||
| } | ||
| Assert.assertEquals( | ||
| getMetric(dvcMetricsRepo, "current_version_number.Gauge", storeNameStringToNameRecord), | ||
| (double) 1); | ||
| } | ||
| } | ||
|
|
||
| @Test(timeOut = TEST_TIMEOUT) | ||
| public void testDVCRequestBasedMetaRepositoryStringToNameRecordVersions() | ||
| throws IOException, ExecutionException, InterruptedException { | ||
|
|
||
| try (DaVinciClient<String, Object> storeClient = | ||
| daVinciClientFactory.getAndStartGenericAvroClient(storeNameStringToNameRecord, daVinciConfig)) { | ||
| storeClient.subscribeAll().get(); | ||
|
|
||
| for (int i = 0; i < TestWriteUtils.countStringToNameRecordSchemas(); i++) { | ||
| Schema schema = TestWriteUtils.getStringToNameRecordSchema(i); | ||
| int currentValueVersion = i + 2; | ||
|
|
||
| int recordCount = currentValueVersion * TestWriteUtils.DEFAULT_USER_DATA_RECORD_COUNT; | ||
| runPushJob( | ||
| storeNameStringToNameRecord, | ||
| TestWriteUtils.writeSimpleAvroFileWithStringToNameRecordSchema( | ||
| getPushJobAvroFileDirectory(storeNameStringToNameRecord), | ||
| schema, | ||
| recordCount)); | ||
|
|
||
| // Perform another push with 200 keys to verify future version ingestion and version swap | ||
| TestUtils.waitForNonDeterministicAssertion(60, TimeUnit.SECONDS, false, () -> { | ||
| Assert.assertEquals( | ||
| getMetric(dvcMetricsRepo, "current_version_number.Gauge", storeNameStringToNameRecord), | ||
| (double) (currentValueVersion)); | ||
| }); | ||
|
|
||
| for (int j = 1; j <= recordCount; j++) { | ||
| Assert.assertEquals( | ||
| storeClient.get(Integer.toString(j)).get().toString(), | ||
| TestWriteUtils.renderNameRecord(schema, j).get(DEFAULT_VALUE_FIELD_PROP).toString()); | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private double getMetric(MetricsRepository metricsRepository, String metricName, String storeName) { | ||
| Metric metric = metricsRepository.getMetric("." + storeName + "--" + metricName); | ||
| assertNotNull(metric, "Expected metric " + metricName + " not found."); | ||
| return metric.value(); | ||
| } | ||
|
|
||
| private File getPushJobAvroFileDirectory(String storeName) { | ||
| if (!pushJobAvroDataDirs.containsKey(storeName)) { | ||
| pushJobAvroDataDirs.put(storeName, getTempDataDirectory()); | ||
| } | ||
|
|
||
| return pushJobAvroDataDirs.get(storeName); | ||
| } | ||
|
|
||
| private void runPushJob(String storeName, Schema schema) { | ||
|
|
||
| ControllerClient controllerClient; | ||
| File dataDir = getPushJobAvroFileDirectory(storeName); | ||
| String dataDirPath = "file:" + dataDir.getAbsolutePath(); | ||
|
|
||
| if (!controllerClients.containsKey(storeName)) { | ||
| // Init store | ||
| controllerClient = IntegrationTestPushUtils.createStoreForJob( | ||
| CLUSTER_NAME, | ||
| schema, | ||
| TestWriteUtils.defaultVPJProps( | ||
| clusterWrapper.getVeniceControllers().get(0).getControllerUrl(), | ||
| dataDirPath, | ||
| storeName)); | ||
| controllerClients.put(storeName, controllerClient); | ||
| } else { | ||
| controllerClient = controllerClients.get(storeName); | ||
|
|
||
| // Add new schema | ||
| Schema valueSchema = schema.getField(DEFAULT_VALUE_FIELD_PROP).schema(); | ||
| SchemaResponse schemaResponse = controllerClient.addValueSchema(storeName, valueSchema.toString()); | ||
| Assert.assertFalse(schemaResponse.isError(), schemaResponse.getError()); | ||
| } | ||
|
|
||
| Properties props = | ||
| TestWriteUtils.defaultVPJProps(controllerClient.getLeaderControllerUrl(), dataDirPath, storeName); | ||
| TestWriteUtils.runPushJob(storeName + "_" + Utils.getUniqueString("push_job"), props); | ||
| } | ||
| } |
Oops, something went wrong.