[docker] Add fast client support to Docker quickstart #2466
+679
−12
There are no files selected for viewing
This file contains hidden or 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 hidden or 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 hidden or 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,181 @@ | ||
| package com.linkedin.venice.fastclient; | ||
|
|
||
| import com.linkedin.avroutil1.compatibility.AvroCompatibilityHelper; | ||
| import com.linkedin.d2.balancer.D2Client; | ||
| import com.linkedin.d2.balancer.D2ClientBuilder; | ||
| import com.linkedin.r2.transport.common.Client; | ||
| import com.linkedin.r2.transport.common.TransportClientFactory; | ||
| import com.linkedin.r2.transport.common.bridge.client.TransportClientAdapter; | ||
| import com.linkedin.r2.transport.http.client.HttpClientFactory; | ||
| import com.linkedin.venice.D2.D2ClientUtils; | ||
| import com.linkedin.venice.client.store.AvroGenericStoreClient; | ||
| import com.linkedin.venice.exceptions.VeniceException; | ||
| import com.linkedin.venice.fastclient.factory.ClientFactory; | ||
| import com.linkedin.venice.fastclient.meta.StoreMetadataFetchMode; | ||
| import java.io.ByteArrayInputStream; | ||
| import java.io.IOException; | ||
| import java.nio.charset.StandardCharsets; | ||
| import java.security.SecureRandom; | ||
| import java.security.cert.X509Certificate; | ||
| import java.util.Collections; | ||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
| import java.util.concurrent.TimeUnit; | ||
| import javax.net.ssl.SSLContext; | ||
| import javax.net.ssl.SSLParameters; | ||
| import javax.net.ssl.TrustManager; | ||
| import javax.net.ssl.X509TrustManager; | ||
| import org.apache.avro.Schema; | ||
| import org.apache.avro.generic.GenericDatumReader; | ||
|
|
||
|
|
||
| /** | ||
| * A CLI tool to query values from a Venice store using the fast client with D2 service discovery. | ||
| * | ||
| * Usage: java -jar venice-client-all.jar {@literal <store_name>} {@literal <key>} {@literal <zk_address>} [--insecure] | ||
| * | ||
| * The optional --insecure flag enables a trust-all SSL context for environments where HTTPS | ||
| * is used with self-signed or untrusted certificates (e.g., Docker quickstart). | ||
| * Without this flag, the tool uses plain HTTP transport only. | ||
| */ | ||
| public class FastClientQueryTool { | ||
| public static void main(String[] args) throws Exception { | ||
| if (args.length < 3) { | ||
| System.out.println("Usage: java -jar venice-client-all.jar <store_name> <key> <zk_address> [--insecure]"); | ||
| System.exit(1); | ||
| } | ||
|
|
||
| String storeName = args[0]; | ||
| String keyString = args[1]; | ||
| String zkAddress = args[2]; | ||
| boolean insecure = args.length > 3 && "--insecure".equals(args[3]); | ||
|
|
||
| TransportClientFactory httpTransport = new HttpClientFactory.Builder().setUsePipelineV2(true).build(); | ||
| D2ClientBuilder d2ClientBuilder = new D2ClientBuilder().setZkHosts(zkAddress) | ||
| .setZkSessionTimeout(5000, TimeUnit.MILLISECONDS) | ||
| .setZkStartupTimeout(5000, TimeUnit.MILLISECONDS) | ||
| .setLbWaitTimeout(5000, TimeUnit.MILLISECONDS) | ||
| .setBasePath("/d2"); | ||
|
|
||
| Map<String, Object> r2Properties; | ||
|
|
||
| if (insecure) { | ||
| // Trust-all SSLContext for environments with self-signed/untrusted certs (non-production use) | ||
| TrustManager[] trustAllCerts = new TrustManager[] { new X509TrustManager() { | ||
| public X509Certificate[] getAcceptedIssuers() { | ||
| return new X509Certificate[0]; | ||
| } | ||
|
|
||
| public void checkClientTrusted(X509Certificate[] certs, String authType) { | ||
| } | ||
|
|
||
| public void checkServerTrusted(X509Certificate[] certs, String authType) { | ||
| } | ||
| } }; | ||
| SSLContext sslContext = SSLContext.getInstance("TLS"); | ||
| sslContext.init(null, trustAllCerts, new SecureRandom()); | ||
| SSLParameters sslParameters = sslContext.getDefaultSSLParameters(); | ||
|
|
||
| Map<String, TransportClientFactory> transportClients = new HashMap<>(); | ||
| transportClients.put("http", httpTransport); | ||
| transportClients.put("https", httpTransport); | ||
|
|
||
| d2ClientBuilder.setSSLContext(sslContext) | ||
| .setSSLParameters(sslParameters) | ||
| .setIsSSLEnabled(true) | ||
| .setClientFactories(transportClients); | ||
|
|
||
| r2Properties = new HashMap<>(); | ||
| r2Properties.put(HttpClientFactory.HTTP_SSL_CONTEXT, sslContext); | ||
| r2Properties.put(HttpClientFactory.HTTP_SSL_PARAMS, sslParameters); | ||
| } else { | ||
| r2Properties = Collections.emptyMap(); | ||
| } | ||
|
|
||
| D2Client d2Client = d2ClientBuilder.build(); | ||
| D2ClientUtils.startClient(d2Client); | ||
|
|
||
| Client r2Client = new TransportClientAdapter(httpTransport.getClient(r2Properties)); | ||
|
|
||
| // Build fast client config | ||
| ClientConfig clientConfig = new ClientConfig.ClientConfigBuilder<>().setStoreName(storeName) | ||
| .setR2Client(r2Client) | ||
| .setD2Client(d2Client) | ||
| .setClusterDiscoveryD2Service("venice-discovery") | ||
| .setStoreMetadataFetchMode(StoreMetadataFetchMode.SERVER_BASED_METADATA) | ||
| .setMetadataRefreshIntervalInSeconds(1) | ||
| .build(); | ||
|
|
||
| try (AvroGenericStoreClient<Object, Object> client = ClientFactory.getAndStartGenericStoreClient(clientConfig)) { | ||
| // Poll until metadata (key schema) is available | ||
| Schema keySchema = null; | ||
| Exception lastException = null; | ||
| long deadline = System.currentTimeMillis() + 30_000; | ||
| while (keySchema == null) { | ||
| if (System.currentTimeMillis() > deadline) { | ||
| String message = "Timed out waiting for metadata to be fetched for store: " + storeName; | ||
| if (lastException != null) { | ||
| throw new VeniceException(message + ". Last error: " + lastException.getMessage(), lastException); | ||
| } | ||
| throw new VeniceException(message); | ||
| } | ||
| try { | ||
| keySchema = client.getKeySchema(); | ||
| } catch (Exception e) { | ||
| // Metadata not yet available, record and retry until timeout | ||
| lastException = e; | ||
| keySchema = null; | ||
| } | ||
| if (keySchema == null) { | ||
| Thread.sleep(200); | ||
| } | ||
| } | ||
sushantmane marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
|
||
| Object key = convertKey(keyString, keySchema); | ||
|
|
||
| Object value = client.get(key).get(15, TimeUnit.SECONDS); | ||
|
|
||
| System.out.println("key-class=" + key.getClass().getCanonicalName()); | ||
| System.out.println("value-class=" + (value == null ? "null" : value.getClass().getCanonicalName())); | ||
| System.out.println("key=" + keyString); | ||
| System.out.println("value=" + (value == null ? "null" : value.toString())); | ||
| } finally { | ||
| D2ClientUtils.shutdownClient(d2Client); | ||
| } | ||
| } | ||
|
|
||
| static Object convertKey(String keyString, Schema keySchema) { | ||
| Object key; | ||
| switch (keySchema.getType()) { | ||
| case INT: | ||
| key = Integer.parseInt(keyString); | ||
| break; | ||
| case LONG: | ||
| key = Long.parseLong(keyString); | ||
| break; | ||
| case FLOAT: | ||
| key = Float.parseFloat(keyString); | ||
| break; | ||
| case DOUBLE: | ||
| key = Double.parseDouble(keyString); | ||
| break; | ||
| case BOOLEAN: | ||
| key = Boolean.parseBoolean(keyString); | ||
| break; | ||
| case STRING: | ||
| key = keyString; | ||
| break; | ||
| default: | ||
| try { | ||
| key = new GenericDatumReader<>(keySchema, keySchema).read( | ||
| null, | ||
| AvroCompatibilityHelper | ||
| .newJsonDecoder(keySchema, new ByteArrayInputStream(keyString.getBytes(StandardCharsets.UTF_8)))); | ||
| } catch (IOException e) { | ||
sushantmane marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| throw new VeniceException("Invalid input key: " + keyString, e); | ||
| } | ||
sushantmane marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| break; | ||
| } | ||
| return key; | ||
| } | ||
| } | ||
This file contains hidden or 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,82 @@ | ||
| package com.linkedin.venice.fastclient; | ||
|
|
||
| import static org.testng.Assert.assertEquals; | ||
| import static org.testng.Assert.assertTrue; | ||
|
|
||
| import com.linkedin.avroutil1.compatibility.AvroCompatibilityHelper; | ||
| import com.linkedin.venice.exceptions.VeniceException; | ||
| import org.apache.avro.Schema; | ||
| import org.testng.annotations.Test; | ||
|
|
||
|
|
||
| public class FastClientQueryToolTest { | ||
| @Test | ||
| public void testConvertKeyString() { | ||
| Schema schema = Schema.create(Schema.Type.STRING); | ||
| Object key = FastClientQueryTool.convertKey("hello", schema); | ||
| assertEquals(key, "hello"); | ||
| } | ||
|
|
||
| @Test | ||
| public void testConvertKeyInt() { | ||
| Schema schema = Schema.create(Schema.Type.INT); | ||
| Object key = FastClientQueryTool.convertKey("42", schema); | ||
| assertEquals(key, 42); | ||
| } | ||
|
|
||
| @Test | ||
| public void testConvertKeyLong() { | ||
| Schema schema = Schema.create(Schema.Type.LONG); | ||
| Object key = FastClientQueryTool.convertKey("123456789", schema); | ||
| assertEquals(key, 123456789L); | ||
| } | ||
|
|
||
| @Test | ||
| public void testConvertKeyFloat() { | ||
| Schema schema = Schema.create(Schema.Type.FLOAT); | ||
| Object key = FastClientQueryTool.convertKey("1.5", schema); | ||
| assertEquals(key, 1.5f); | ||
| } | ||
|
|
||
| @Test | ||
| public void testConvertKeyDouble() { | ||
| Schema schema = Schema.create(Schema.Type.DOUBLE); | ||
| Object key = FastClientQueryTool.convertKey("1.5", schema); | ||
| assertEquals(key, 1.5); | ||
| } | ||
|
|
||
| @Test | ||
| public void testConvertKeyBoolean() { | ||
| Schema schema = Schema.create(Schema.Type.BOOLEAN); | ||
| Object key = FastClientQueryTool.convertKey("true", schema); | ||
| assertEquals(key, true); | ||
|
|
||
| key = FastClientQueryTool.convertKey("false", schema); | ||
| assertEquals(key, false); | ||
| } | ||
|
|
||
| @Test(expectedExceptions = NumberFormatException.class) | ||
| public void testConvertKeyIntWithInvalidInput() { | ||
| Schema schema = Schema.create(Schema.Type.INT); | ||
| FastClientQueryTool.convertKey("not_a_number", schema); | ||
| } | ||
|
|
||
| @Test(expectedExceptions = VeniceException.class) | ||
| public void testConvertKeyComplexSchemaWithInvalidJson() { | ||
| Schema schema = Schema.createRecord("TestRecord", null, "test", false); | ||
| schema.setFields( | ||
| java.util.Collections.singletonList( | ||
| AvroCompatibilityHelper.createSchemaField("field1", Schema.create(Schema.Type.STRING), null, null))); | ||
| FastClientQueryTool.convertKey("not_valid_json", schema); | ||
| } | ||
|
|
||
| @Test | ||
| public void testConvertKeyComplexSchemaWithValidJson() { | ||
| Schema schema = Schema.createRecord("TestRecord", null, "test", false); | ||
| schema.setFields( | ||
| java.util.Collections.singletonList( | ||
| AvroCompatibilityHelper.createSchemaField("field1", Schema.create(Schema.Type.STRING), null, null))); | ||
| Object key = FastClientQueryTool.convertKey("{\"field1\": \"value1\"}", schema); | ||
| assertTrue(key.toString().contains("value1")); | ||
| } | ||
| } |
This file contains hidden or 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 hidden or 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 hidden or 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,10 @@ | ||
| #!/bin/bash | ||
|
|
||
| if [ "$#" -lt 2 ]; then | ||
| echo "Usage: $0 <storeName> <key>" >&2 | ||
| exit 1 | ||
| fi | ||
|
|
||
| storeName=$1 | ||
| key=$2 | ||
| java -jar /opt/venice/bin/venice-client-all.jar "$storeName" "$key" zookeeper:2181 2>/dev/null | ||
sushantmane marked this conversation as resolved.
Show resolved
Hide resolved
|
||
This file contains hidden or 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 hidden or 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
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.