Skip to content

Commit d37e6a9

Browse files
ishwarya-citro“ishwarya-personal”
ishwarya-citro
and
“ishwarya-personal”
authored
[dvc] dvc node blob discovery (#980)
* dvc blob discovery - router changes --------- Co-authored-by: “ishwarya-personal” <“[email protected]”>
1 parent e382726 commit d37e6a9

File tree

11 files changed

+717
-10
lines changed

11 files changed

+717
-10
lines changed

‎internal/venice-client-common/src/main/java/com/linkedin/venice/routerapi/BlobDiscoveryResponse.java

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,17 @@
1+
package com.linkedin.venice.routerapi;
2+
3+
import com.linkedin.venice.controllerapi.ControllerResponse;
4+
import java.util.List;
5+
6+
7+
public class BlobDiscoveryResponse extends ControllerResponse {
8+
private List<String> liveNodeHostNames;
9+
10+
public void setLiveNodeNames(List<String> liveNodeHostNames) {
11+
this.liveNodeHostNames = liveNodeHostNames;
12+
}
13+
14+
public List<String> getLiveNodeHostNames() {
15+
return liveNodeHostNames;
16+
}
17+
}

‎internal/venice-common/src/main/java/com/linkedin/venice/ConfigConstants.java

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package com.linkedin.venice;
22

3+
import java.util.concurrent.TimeUnit;
34
import org.apache.logging.log4j.LogManager;
45
import org.apache.logging.log4j.Logger;
56

@@ -12,6 +13,9 @@ public class ConfigConstants {
1213

1314
public static final int UNSPECIFIED_REPLICATION_METADATA_VERSION = -1;
1415

16+
public static final long DEFAULT_PUSH_STATUS_STORE_HEARTBEAT_EXPIRATION_TIME_IN_SECONDS =
17+
TimeUnit.MINUTES.toSeconds(10);
18+
1519
/**
1620
* End of controller config default value
1721
*/

‎internal/venice-common/src/main/java/com/linkedin/venice/controllerapi/ControllerApiConstants.java

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,8 @@ public class ControllerApiConstants {
88
public static final String BATCH_JOB_HEARTBEAT_ENABLED = "batch_job_heartbeat_enabled";
99

1010
public static final String NAME = "store_name";
11+
public static final String STORE_PARTITION = "store_partition";
12+
public static final String STORE_VERSION = "store_version";
1113
public static final String OWNER = "owner";
1214
public static final String FABRIC = "fabric";
1315
public static final String FABRIC_A = "fabric_a";

‎internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/router/TestBlobDiscovery.java

Lines changed: 235 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,235 @@
1+
package com.linkedin.venice.router;
2+
3+
import static com.linkedin.venice.ConfigKeys.CLIENT_SYSTEM_STORE_REPOSITORY_REFRESH_INTERVAL_SECONDS;
4+
import static com.linkedin.venice.ConfigKeys.CLIENT_USE_SYSTEM_STORE_REPOSITORY;
5+
import static com.linkedin.venice.ConfigKeys.DATA_BASE_PATH;
6+
import static com.linkedin.venice.ConfigKeys.OFFLINE_JOB_START_TIMEOUT_MS;
7+
import static com.linkedin.venice.ConfigKeys.PERSISTENCE_TYPE;
8+
import static com.linkedin.venice.router.api.VenicePathParser.TYPE_BLOB_DISCOVERY;
9+
import static org.testng.Assert.assertFalse;
10+
import static org.testng.Assert.fail;
11+
12+
import com.fasterxml.jackson.databind.ObjectMapper;
13+
import com.linkedin.d2.balancer.D2Client;
14+
import com.linkedin.davinci.client.DaVinciClient;
15+
import com.linkedin.davinci.client.DaVinciConfig;
16+
import com.linkedin.davinci.client.factory.CachingDaVinciClientFactory;
17+
import com.linkedin.venice.D2.D2ClientUtils;
18+
import com.linkedin.venice.common.VeniceSystemStoreType;
19+
import com.linkedin.venice.common.VeniceSystemStoreUtils;
20+
import com.linkedin.venice.controllerapi.ControllerClient;
21+
import com.linkedin.venice.controllerapi.UpdateStoreQueryParams;
22+
import com.linkedin.venice.controllerapi.VersionCreationResponse;
23+
import com.linkedin.venice.exceptions.VeniceException;
24+
import com.linkedin.venice.integration.utils.D2TestUtils;
25+
import com.linkedin.venice.integration.utils.PubSubBrokerWrapper;
26+
import com.linkedin.venice.integration.utils.ServiceFactory;
27+
import com.linkedin.venice.integration.utils.VeniceClusterWrapper;
28+
import com.linkedin.venice.integration.utils.VeniceControllerWrapper;
29+
import com.linkedin.venice.integration.utils.VeniceMultiClusterWrapper;
30+
import com.linkedin.venice.integration.utils.VeniceRouterWrapper;
31+
import com.linkedin.venice.integration.utils.VeniceTwoLayerMultiRegionMultiClusterWrapper;
32+
import com.linkedin.venice.meta.PersistenceType;
33+
import com.linkedin.venice.meta.Version;
34+
import com.linkedin.venice.pubsub.PubSubProducerAdapterFactory;
35+
import com.linkedin.venice.routerapi.BlobDiscoveryResponse;
36+
import com.linkedin.venice.utils.ObjectMapperFactory;
37+
import com.linkedin.venice.utils.PropertyBuilder;
38+
import com.linkedin.venice.utils.TestUtils;
39+
import com.linkedin.venice.utils.Time;
40+
import com.linkedin.venice.utils.Utils;
41+
import com.linkedin.venice.utils.VeniceProperties;
42+
import io.tehuti.metrics.MetricsRepository;
43+
import java.io.InputStream;
44+
import java.nio.charset.Charset;
45+
import java.util.AbstractMap;
46+
import java.util.ArrayList;
47+
import java.util.List;
48+
import java.util.Map;
49+
import java.util.Optional;
50+
import java.util.Properties;
51+
import java.util.concurrent.ExecutionException;
52+
import java.util.concurrent.TimeUnit;
53+
import java.util.stream.Collectors;
54+
import java.util.stream.IntStream;
55+
import org.apache.commons.httpclient.HttpStatus;
56+
import org.apache.commons.io.IOUtils;
57+
import org.apache.http.HttpResponse;
58+
import org.apache.http.client.config.RequestConfig;
59+
import org.apache.http.client.methods.HttpGet;
60+
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
61+
import org.apache.http.impl.nio.client.HttpAsyncClients;
62+
import org.testng.Assert;
63+
import org.testng.annotations.AfterTest;
64+
import org.testng.annotations.BeforeClass;
65+
import org.testng.annotations.Test;
66+
67+
68+
public class TestBlobDiscovery {
69+
private static final String INT_KEY_SCHEMA = "\"int\"";
70+
private static final String INT_VALUE_SCHEMA = "\"int\"";
71+
String clusterName;
72+
String storeName;
73+
private VeniceMultiClusterWrapper multiClusterVenice;
74+
D2Client daVinciD2;
75+
76+
/**
77+
* Set up a multi-cluster Venice environment with meta system store enabled Venice stores.
78+
*/
79+
80+
@BeforeClass(alwaysRun = true)
81+
public void setUp() {
82+
Utils.thisIsLocalhost();
83+
84+
Properties parentControllerProps = new Properties();
85+
parentControllerProps.put(OFFLINE_JOB_START_TIMEOUT_MS, "180000");
86+
87+
VeniceTwoLayerMultiRegionMultiClusterWrapper multiRegionMultiClusterWrapper =
88+
ServiceFactory.getVeniceTwoLayerMultiRegionMultiClusterWrapper(
89+
1,
90+
2,
91+
1,
92+
1,
93+
3,
94+
1,
95+
3,
96+
Optional.of(parentControllerProps),
97+
Optional.empty(),
98+
Optional.empty(),
99+
false);
100+
101+
multiClusterVenice = multiRegionMultiClusterWrapper.getChildRegions().get(0);
102+
String[] clusterNames = multiClusterVenice.getClusterNames();
103+
String parentControllerURLs = multiRegionMultiClusterWrapper.getParentControllers()
104+
.stream()
105+
.map(VeniceControllerWrapper::getControllerUrl)
106+
.collect(Collectors.joining(","));
107+
108+
for (String cluster: clusterNames) {
109+
try (ControllerClient controllerClient =
110+
new ControllerClient(cluster, multiClusterVenice.getControllerConnectString())) {
111+
// Verify the participant store is up and running in child region
112+
String participantStoreName = VeniceSystemStoreUtils.getParticipantStoreNameForCluster(cluster);
113+
TestUtils.waitForNonDeterministicPushCompletion(
114+
Version.composeKafkaTopic(participantStoreName, 1),
115+
controllerClient,
116+
5,
117+
TimeUnit.MINUTES);
118+
}
119+
}
120+
121+
clusterName = clusterNames[0];
122+
storeName = Utils.getUniqueString("test-store");
123+
124+
List<PubSubBrokerWrapper> pubSubBrokerWrappers = multiClusterVenice.getClusters()
125+
.values()
126+
.stream()
127+
.map(VeniceClusterWrapper::getPubSubBrokerWrapper)
128+
.collect(Collectors.toList());
129+
Map<String, String> additionalPubSubProperties =
130+
PubSubBrokerWrapper.getBrokerDetailsForClients(pubSubBrokerWrappers);
131+
132+
try (ControllerClient parentControllerClient = new ControllerClient(clusterName, parentControllerURLs)) {
133+
assertFalse(
134+
parentControllerClient.createNewStore(storeName, "venice-test", INT_KEY_SCHEMA, INT_VALUE_SCHEMA).isError());
135+
136+
PubSubProducerAdapterFactory pubSubProducerAdapterFactory = multiClusterVenice.getClusters()
137+
.get(clusterName)
138+
.getPubSubBrokerWrapper()
139+
.getPubSubClientsFactory()
140+
.getProducerAdapterFactory();
141+
142+
VersionCreationResponse response = TestUtils.createVersionWithBatchData(
143+
parentControllerClient,
144+
storeName,
145+
INT_KEY_SCHEMA,
146+
INT_VALUE_SCHEMA,
147+
IntStream.range(0, 10).mapToObj(i -> new AbstractMap.SimpleEntry<>(i, 0)),
148+
pubSubProducerAdapterFactory,
149+
additionalPubSubProperties);
150+
151+
// Verify the data can be ingested by classical Venice before proceeding.
152+
TestUtils.waitForNonDeterministicPushCompletion(
153+
response.getKafkaTopic(),
154+
parentControllerClient,
155+
30,
156+
TimeUnit.SECONDS);
157+
158+
makeSureSystemStoresAreOnline(parentControllerClient, storeName);
159+
multiClusterVenice.getClusters().get(clusterName).refreshAllRouterMetaData();
160+
}
161+
162+
VeniceProperties backendConfig =
163+
new PropertyBuilder().put(DATA_BASE_PATH, Utils.getTempDataDirectory().getAbsolutePath())
164+
.put(PERSISTENCE_TYPE, PersistenceType.ROCKS_DB)
165+
.put(CLIENT_USE_SYSTEM_STORE_REPOSITORY, true)
166+
.put(CLIENT_SYSTEM_STORE_REPOSITORY_REFRESH_INTERVAL_SECONDS, 1)
167+
.build();
168+
169+
DaVinciConfig daVinciConfig = new DaVinciConfig();
170+
daVinciD2 = D2TestUtils.getAndStartD2Client(multiClusterVenice.getZkServerWrapper().getAddress());
171+
172+
try (CachingDaVinciClientFactory factory = new CachingDaVinciClientFactory(
173+
daVinciD2,
174+
VeniceRouterWrapper.CLUSTER_DISCOVERY_D2_SERVICE_NAME,
175+
new MetricsRepository(),
176+
backendConfig)) {
177+
List<DaVinciClient<Integer, Object>> clients = new ArrayList<>();
178+
DaVinciClient<Integer, Object> client = factory.getAndStartGenericAvroClient(storeName, daVinciConfig);
179+
client.subscribeAll().get();
180+
clients.add(client);
181+
} catch (ExecutionException | InterruptedException e) {
182+
throw new VeniceException(e);
183+
}
184+
}
185+
186+
@AfterTest
187+
public void tearDown() {
188+
D2ClientUtils.shutdownClient(daVinciD2);
189+
}
190+
191+
@Test(timeOut = 60 * Time.MS_PER_SECOND)
192+
public void testBlobDiscovery() throws Exception {
193+
VeniceClusterWrapper veniceClusterWrapper = multiClusterVenice.getClusters().get(clusterName);
194+
TestUtils.waitForNonDeterministicAssertion(2, TimeUnit.MINUTES, true, () -> {
195+
veniceClusterWrapper.updateStore(storeName, new UpdateStoreQueryParams().setBlobTransferEnabled(true));
196+
});
197+
198+
String routerURL = veniceClusterWrapper.getRandomRouterURL();
199+
200+
try (CloseableHttpAsyncClient client = HttpAsyncClients.custom()
201+
.setDefaultRequestConfig(RequestConfig.custom().setSocketTimeout(4000).build())
202+
.build()) {
203+
client.start();
204+
205+
String uri =
206+
routerURL + "/" + TYPE_BLOB_DISCOVERY + "?store_name=" + storeName + "&store_version=1&store_partition=1";
207+
HttpGet routerRequest = new HttpGet(uri);
208+
HttpResponse response = client.execute(routerRequest, null).get();
209+
String responseBody;
210+
try (InputStream bodyStream = response.getEntity().getContent()) {
211+
responseBody = IOUtils.toString(bodyStream, Charset.defaultCharset());
212+
}
213+
Assert.assertEquals(
214+
response.getStatusLine().getStatusCode(),
215+
HttpStatus.SC_OK,
216+
"Failed to get resource state for " + storeName + ". Response: " + responseBody);
217+
ObjectMapper mapper = ObjectMapperFactory.getInstance();
218+
BlobDiscoveryResponse blobDiscoveryResponse =
219+
mapper.readValue(responseBody.getBytes(), BlobDiscoveryResponse.class);
220+
// TODO: add another testcase to retrieve >= 1 live nodes
221+
Assert.assertEquals(blobDiscoveryResponse.getLiveNodeHostNames().size(), 0);
222+
} catch (Exception e) {
223+
fail("Unexpected exception", e);
224+
}
225+
}
226+
227+
private void makeSureSystemStoresAreOnline(ControllerClient controllerClient, String storeName) {
228+
String metaSystemStoreTopic =
229+
Version.composeKafkaTopic(VeniceSystemStoreType.META_STORE.getSystemStoreName(storeName), 1);
230+
TestUtils.waitForNonDeterministicPushCompletion(metaSystemStoreTopic, controllerClient, 30, TimeUnit.SECONDS);
231+
String daVinciPushStatusStore =
232+
Version.composeKafkaTopic(VeniceSystemStoreType.DAVINCI_PUSH_STATUS_STORE.getSystemStoreName(storeName), 1);
233+
TestUtils.waitForNonDeterministicPushCompletion(daVinciPushStatusStore, controllerClient, 30, TimeUnit.SECONDS);
234+
}
235+
}

‎internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/router/TestRead.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -143,7 +143,7 @@ public void setUp() throws VeniceClientException, ExecutionException, Interrupte
143143
// To trigger long-tail retry
144144
extraProperties.put(ConfigKeys.ROUTER_LONG_TAIL_RETRY_FOR_SINGLE_GET_THRESHOLD_MS, 1);
145145
extraProperties.put(ConfigKeys.ROUTER_MAX_KEY_COUNT_IN_MULTIGET_REQ, MAX_KEY_LIMIT); // 20 keys at most in a
146-
// batch-get request
146+
// batch-get request
147147
extraProperties.put(ConfigKeys.ROUTER_LONG_TAIL_RETRY_FOR_BATCH_GET_THRESHOLD_MS, "1-:1");
148148
extraProperties.put(ConfigKeys.ROUTER_SMART_LONG_TAIL_RETRY_ENABLED, false);
149149
extraProperties.put(ConfigKeys.ROUTER_STORAGE_NODE_CLIENT_TYPE, getStorageNodeClientType());

‎services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceControllerConfig.java

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package com.linkedin.venice.controller;
22

3+
import static com.linkedin.venice.ConfigConstants.DEFAULT_PUSH_STATUS_STORE_HEARTBEAT_EXPIRATION_TIME_IN_SECONDS;
34
import static com.linkedin.venice.ConfigKeys.ACTIVE_ACTIVE_ENABLED_ON_CONTROLLER;
45
import static com.linkedin.venice.ConfigKeys.ACTIVE_ACTIVE_REAL_TIME_SOURCE_FABRIC_LIST;
56
import static com.linkedin.venice.ConfigKeys.ADMIN_CHECK_READ_METHOD_FOR_KAFKA;
@@ -510,8 +511,9 @@ public VeniceControllerConfig(VeniceProperties props) {
510511
*/
511512
this.zkSharedMetaSystemSchemaStoreAutoCreationEnabled =
512513
props.getBoolean(CONTROLLER_ZK_SHARED_META_SYSTEM_SCHEMA_STORE_AUTO_CREATION_ENABLED, false);
513-
this.pushStatusStoreHeartbeatExpirationTimeInSeconds =
514-
props.getLong(PUSH_STATUS_STORE_HEARTBEAT_EXPIRATION_TIME_IN_SECONDS, TimeUnit.MINUTES.toSeconds(10));
514+
this.pushStatusStoreHeartbeatExpirationTimeInSeconds = props.getLong(
515+
PUSH_STATUS_STORE_HEARTBEAT_EXPIRATION_TIME_IN_SECONDS,
516+
DEFAULT_PUSH_STATUS_STORE_HEARTBEAT_EXPIRATION_TIME_IN_SECONDS);
515517
this.isDaVinciPushStatusStoreEnabled = props.getBoolean(PUSH_STATUS_STORE_ENABLED, false);
516518
this.daVinciPushStatusScanEnabled =
517519
props.getBoolean(DAVINCI_PUSH_STATUS_SCAN_ENABLED, true) && isDaVinciPushStatusStoreEnabled;

0 commit comments

Comments
 (0)