diff --git a/.prow/scripts/test-end-to-end-batch.sh b/.prow/scripts/test-end-to-end-batch.sh index 4d1b1d2ecd7..268bd248c17 100755 --- a/.prow/scripts/test-end-to-end-batch.sh +++ b/.prow/scripts/test-end-to-end-batch.sh @@ -3,11 +3,12 @@ set -e set -o pipefail -if ! cat /etc/*release | grep -q stretch; then - echo ${BASH_SOURCE} only supports Debian stretch. - echo Please change your operating system to use this script. - exit 1 -fi +test -z ${GOOGLE_APPLICATION_CREDENTIALS} && GOOGLE_APPLICATION_CREDENTIALS="/etc/service-account/service-account.json" +test -z ${SKIP_BUILD_JARS} && SKIP_BUILD_JARS="false" +test -z ${GOOGLE_CLOUD_PROJECT} && GOOGLE_CLOUD_PROJECT="kf-feast" +test -z ${TEMP_BUCKET} && TEMP_BUCKET="feast-templocation-kf-feast" +test -z ${JOBS_STAGING_LOCATION} && JOBS_STAGING_LOCATION="gs://${TEMP_BUCKET}/staging-location" +test -z ${JAR_VERSION_SUFFIX} && JAR_VERSION_SUFFIX="-SNAPSHOT" echo " This script will run end-to-end tests for Feast Core and Batch Serving. @@ -16,10 +17,13 @@ This script will run end-to-end tests for Feast Core and Batch Serving. 2. Install Redis as the job store for Feast Batch Serving. 4. Install Postgres for persisting Feast metadata. 5. Install Kafka and Zookeeper as the Source in Feast. -6. Install Python 3.7.4, Feast Python SDK and run end-to-end tests from +6. Install Python 3.7.4, Feast Python SDK and run end-to-end tests from tests/e2e via pytest. " +apt-get -qq update +apt-get -y install wget netcat kafkacat + echo " ============================================================ @@ -31,8 +35,8 @@ if [[ ! $(command -v gsutil) ]]; then . "${CURRENT_DIR}"/install-google-cloud-sdk.sh fi -export GOOGLE_APPLICATION_CREDENTIALS=/etc/service-account/service-account.json -gcloud auth activate-service-account --key-file /etc/service-account/service-account.json +export GOOGLE_APPLICATION_CREDENTIALS +gcloud auth activate-service-account --key-file ${GOOGLE_APPLICATION_CREDENTIALS} @@ -41,10 +45,9 @@ echo " Installing Redis at localhost:6379 ============================================================ " -apt-get -qq update # Allow starting serving in this Maven Docker image. Default set to not allowed. echo "exit 0" > /usr/sbin/policy-rc.d -apt-get -y install redis-server wget > /var/log/redis.install.log +apt-get -y install redis-server > /var/log/redis.install.log redis-server --daemonize yes redis-cli ping @@ -73,24 +76,32 @@ Installing Kafka at localhost:9092 wget -qO- https://www-eu.apache.org/dist/kafka/2.3.0/kafka_2.12-2.3.0.tgz | tar xz mv kafka_2.12-2.3.0/ /tmp/kafka nohup /tmp/kafka/bin/zookeeper-server-start.sh /tmp/kafka/config/zookeeper.properties &> /var/log/zookeeper.log 2>&1 & -sleep 10 +sleep 5 tail -n10 /var/log/zookeeper.log nohup /tmp/kafka/bin/kafka-server-start.sh /tmp/kafka/config/server.properties &> /var/log/kafka.log 2>&1 & -sleep 30 +sleep 20 tail -n10 /var/log/kafka.log - -echo " -============================================================ -Building jars for Feast -============================================================ -" - -.prow/scripts/download-maven-cache.sh \ - --archive-uri gs://feast-templocation-kf-feast/.m2.2019-10-24.tar \ - --output-dir /root/ - -# Build jars for Feast -mvn --quiet --batch-mode --define skipTests=true clean package +kafkacat -b localhost:9092 -L + +if [[ ${SKIP_BUILD_JARS} != "true" ]]; then + echo " + ============================================================ + Building jars for Feast + ============================================================ + " + + .prow/scripts/download-maven-cache.sh \ + --archive-uri gs://feast-templocation-kf-feast/.m2.2019-10-24.tar \ + --output-dir /root/ + + # Build jars for Feast + mvn --quiet --batch-mode --define skipTests=true clean package + + ls -lh core/target/*jar + ls -lh serving/target/*jar +else + echo "[DEBUG] Skipping building jars" +fi echo " ============================================================ @@ -142,11 +153,13 @@ management: enabled: false EOF -nohup java -jar core/target/feast-core-*-SNAPSHOT.jar \ +nohup java -jar core/target/feast-core-*${JAR_VERSION_SUFFIX}.jar \ --spring.config.location=file:///tmp/core.application.yml \ &> /var/log/feast-core.log & sleep 35 tail -n10 /var/log/feast-core.log +nc -w2 localhost 6565 < /dev/null + echo " ============================================================ Starting Feast Warehouse Serving @@ -155,18 +168,18 @@ Starting Feast Warehouse Serving DATASET_NAME=feast_$(date +%s) -bq --location=US --project_id=kf-feast mk \ +bq --location=US --project_id=${GOOGLE_CLOUD_PROJECT} mk \ --dataset \ --default_table_expiration 86400 \ - kf-feast:$DATASET_NAME + ${GOOGLE_CLOUD_PROJECT}:${DATASET_NAME} # Start Feast Online Serving in background cat < /tmp/serving.store.bigquery.yml name: warehouse type: BIGQUERY bigquery_config: - projectId: kf-feast - datasetId: $DATASET_NAME + projectId: ${GOOGLE_CLOUD_PROJECT} + datasetId: ${DATASET_NAME} subscriptions: - name: "*" version: "*" @@ -183,26 +196,29 @@ feast: store: config-path: /tmp/serving.store.bigquery.yml jobs: - staging-location: gs://feast-templocation-kf-feast/staging-location + staging-location: ${JOBS_STAGING_LOCATION} store-type: REDIS bigquery-initial-retry-delay-secs: 1 bigquery-total-timeout-secs: 900 store-options: - host: $REMOTE_HOST + host: localhost port: 6379 grpc: port: 6566 enable-reflection: true + spring: main: web-environment: false + EOF -nohup java -jar serving/target/feast-serving-*-SNAPSHOT.jar \ +nohup java -jar serving/target/feast-serving-*${JAR_VERSION_SUFFIX}.jar \ --spring.config.location=file:///tmp/serving.warehouse.application.yml \ &> /var/log/feast-serving-warehouse.log & sleep 15 tail -n100 /var/log/feast-serving-warehouse.log +nc -w2 localhost 6566 < /dev/null echo " ============================================================ @@ -232,16 +248,13 @@ ORIGINAL_DIR=$(pwd) cd tests/e2e set +e -pytest bq-batch-retrieval.py --junitxml=${LOGS_ARTIFACT_PATH}/python-sdk-test-report.xml +pytest bq-batch-retrieval.py --gcs_path "gs://${TEMP_BUCKET}/" --junitxml=${LOGS_ARTIFACT_PATH}/python-sdk-test-report.xml TEST_EXIT_CODE=$? if [[ ${TEST_EXIT_CODE} != 0 ]]; then echo "[DEBUG] Printing logs" ls -ltrh /var/log/feast* cat /var/log/feast-serving-warehouse.log /var/log/feast-core.log - - echo "[DEBUG] Printing Python packages list" - pip list fi cd ${ORIGINAL_DIR} @@ -253,4 +266,4 @@ Cleaning up ============================================================ " -bq rm -r -f kf-feast:$DATASET_NAME +bq rm -r -f ${GOOGLE_CLOUD_PROJECT}:${DATASET_NAME} diff --git a/.prow/scripts/test-end-to-end.sh b/.prow/scripts/test-end-to-end.sh index b9d7fa90882..c436d2f6905 100755 --- a/.prow/scripts/test-end-to-end.sh +++ b/.prow/scripts/test-end-to-end.sh @@ -3,11 +3,12 @@ set -e set -o pipefail -if ! cat /etc/*release | grep -q stretch; then - echo ${BASH_SOURCE} only supports Debian stretch. - echo Please change your operating system to use this script. - exit 1 -fi +test -z ${GOOGLE_APPLICATION_CREDENTIALS} && GOOGLE_APPLICATION_CREDENTIALS="/etc/service-account/service-account.json" +test -z ${SKIP_BUILD_JARS} && SKIP_BUILD_JARS="false" +test -z ${GOOGLE_CLOUD_PROJECT} && GOOGLE_CLOUD_PROJECT="kf-feast" +test -z ${TEMP_BUCKET} && TEMP_BUCKET="feast-templocation-kf-feast" +test -z ${JOBS_STAGING_LOCATION} && JOBS_STAGING_LOCATION="gs://${TEMP_BUCKET}/staging-location" +test -z ${JAR_VERSION_SUFFIX} && JAR_VERSION_SUFFIX="-SNAPSHOT" echo " This script will run end-to-end tests for Feast Core and Online Serving. @@ -15,7 +16,7 @@ This script will run end-to-end tests for Feast Core and Online Serving. 1. Install Redis as the store for Feast Online Serving. 2. Install Postgres for persisting Feast metadata. 3. Install Kafka and Zookeeper as the Source in Feast. -4. Install Python 3.7.4, Feast Python SDK and run end-to-end tests from +4. Install Python 3.7.4, Feast Python SDK and run end-to-end tests from tests/e2e via pytest. " @@ -27,7 +28,6 @@ echo " Installing Redis at localhost:6379 ============================================================ " - # Allow starting serving in this Maven Docker image. Default set to not allowed. echo "exit 0" > /usr/sbin/policy-rc.d apt-get -y install redis-server > /var/log/redis.install.log @@ -67,24 +67,24 @@ tail -n10 /var/log/kafka.log kafkacat -b localhost:9092 -L if [[ ${SKIP_BUILD_JARS} != "true" ]]; then -echo " -============================================================ -Building jars for Feast -============================================================ -" + echo " + ============================================================ + Building jars for Feast + ============================================================ + " -.prow/scripts/download-maven-cache.sh \ - --archive-uri gs://feast-templocation-kf-feast/.m2.2019-10-24.tar \ - --output-dir /root/ + .prow/scripts/download-maven-cache.sh \ + --archive-uri gs://feast-templocation-kf-feast/.m2.2019-10-24.tar \ + --output-dir /root/ -# Build jars for Feast -mvn --quiet --batch-mode --define skipTests=true clean package + # Build jars for Feast + mvn --quiet --batch-mode --define skipTests=true clean package -ls -lh core/target/*jar -ls -lh serving/target/*jar -else - echo "[DEBUG] Skipping building jars" -fi + ls -lh core/target/*jar + ls -lh serving/target/*jar + else + echo "[DEBUG] Skipping building jars" + fi echo " ============================================================ @@ -122,7 +122,6 @@ spring: event.merge.entity_copy_observer: allow hibernate.naming.physical-strategy=org.hibernate.boot.model.naming: PhysicalNamingStrategyStandardImpl hibernate.ddl-auto: update - datasource: url: jdbc:postgresql://localhost:5432/postgres username: postgres @@ -137,7 +136,7 @@ management: enabled: false EOF -nohup java -jar core/target/feast-core-*-SNAPSHOT.jar \ +nohup java -jar core/target/feast-core-*${JAR_VERSION_SUFFIX}.jar \ --spring.config.location=file:///tmp/core.application.yml \ &> /var/log/feast-core.log & sleep 35 @@ -167,17 +166,14 @@ feast: version: 0.3 core-host: localhost core-grpc-port: 6565 - tracing: enabled: false - store: config-path: /tmp/serving.store.redis.yml redis-pool-max-size: 128 redis-pool-max-idle: 16 - jobs: - staging-location: gs://feast-templocation-kf-feast/staging-location + staging-location: ${JOBS_STAGING_LOCATION} store-type: store-options: {} @@ -191,11 +187,11 @@ spring: EOF -nohup java -jar serving/target/feast-serving-*-SNAPSHOT.jar \ +nohup java -jar serving/target/feast-serving-*${JAR_VERSION_SUFFIX}.jar \ --spring.config.location=file:///tmp/serving.online.application.yml \ &> /var/log/feast-serving-online.log & sleep 15 -tail -n10 /var/log/feast-serving-online.log +tail -n100 /var/log/feast-serving-online.log nc -w2 localhost 6566 < /dev/null echo " @@ -233,9 +229,6 @@ if [[ ${TEST_EXIT_CODE} != 0 ]]; then echo "[DEBUG] Printing logs" ls -ltrh /var/log/feast* cat /var/log/feast-serving-online.log /var/log/feast-core.log - - echo "[DEBUG] Printing Python packages list" - pip list fi cd ${ORIGINAL_DIR} diff --git a/CHANGELOG.md b/CHANGELOG.md index 7758ae3fb97..4276858ca0a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,18 @@ # Changelog +## [v0.4.7](https://github.com/gojek/feast/tree/v0.4.7) (2020-03-17) + +[Full Changelog](https://github.com/gojek/feast/compare/v0.4.6...v0.4.7) + +**Merged pull requests:** +- Add log4j-web jar to core and serving. [\#498](https://github.com/gojek/feast/pull/498) ([Yanson](https://github.com/Yanson)) +- Clear all the futures when sync is called. [\#501](https://github.com/gojek/feast/pull/501) ([lavkesh](https://github.com/lavkesh)) +- Encode feature row before storing in Redis [\#530](https://github.com/gojek/feast/pull/530) ([khorshuheng](https://github.com/khorshuheng)) +- Remove transaction when listing projects [\#522](https://github.com/gojek/feast/pull/522) ([davidheryanto](https://github.com/davidheryanto)) +- Remove unused ingestion deps [\#520](https://github.com/gojek/feast/pull/520) ([ches](https://github.com/ches)) +- Parameterize end to end test scripts. [\#433](https://github.com/gojek/feast/pull/433) ([Yanson](https://github.com/Yanson)) +- Replacing Jedis With Lettuce in ingestion and serving [\#485](https://github.com/gojek/feast/pull/485) ([lavkesh](https://github.com/lavkesh)) + ## [v0.4.6](https://github.com/gojek/feast/tree/v0.4.6) (2020-02-26) [Full Changelog](https://github.com/gojek/feast/compare/v0.4.5...v0.4.6) diff --git a/core/pom.xml b/core/pom.xml index e1567ae8fe3..fdee512db55 100644 --- a/core/pom.xml +++ b/core/pom.xml @@ -72,6 +72,10 @@ org.springframework.boot spring-boot-starter-log4j2 + + org.apache.logging.log4j + log4j-web + io.github.lognet diff --git a/core/src/main/java/feast/core/service/AccessManagementService.java b/core/src/main/java/feast/core/service/AccessManagementService.java index 6f627df33d6..df92750e94f 100644 --- a/core/src/main/java/feast/core/service/AccessManagementService.java +++ b/core/src/main/java/feast/core/service/AccessManagementService.java @@ -71,7 +71,6 @@ public void archiveProject(String name) { * * @return List of active projects */ - @Transactional public List listProjects() { return projectRepository.findAllByArchivedIsFalse(); } diff --git a/datatypes/java/README.md b/datatypes/java/README.md index 28b693786c8..a41c1ff04dd 100644 --- a/datatypes/java/README.md +++ b/datatypes/java/README.md @@ -16,7 +16,7 @@ Dependency Coordinates dev.feast datatypes-java - 0.4.6-SNAPSHOT + 0.4.7-SNAPSHOT ``` diff --git a/datatypes/java/pom.xml b/datatypes/java/pom.xml index a6dfa8e345a..415d204293a 100644 --- a/datatypes/java/pom.xml +++ b/datatypes/java/pom.xml @@ -37,6 +37,17 @@ + + org.apache.maven.plugins + maven-dependency-plugin + + + + javax.annotation + + + + org.xolstice.maven.plugins protobuf-maven-plugin @@ -64,9 +75,36 @@ + + + com.google.guava + guava + + + com.google.protobuf + protobuf-java + + + + io.grpc + grpc-core + + + io.grpc + grpc-protobuf + io.grpc grpc-services + + io.grpc + grpc-stub + + + + javax.annotation + javax.annotation-api + diff --git a/infra/charts/feast/Chart.yaml b/infra/charts/feast/Chart.yaml index 35a02a7f894..433ac009f8b 100644 --- a/infra/charts/feast/Chart.yaml +++ b/infra/charts/feast/Chart.yaml @@ -1,4 +1,4 @@ apiVersion: v1 description: A Helm chart to install Feast on kubernetes name: feast -version: 0.4.6 +version: 0.4.7 diff --git a/infra/charts/feast/README.md b/infra/charts/feast/README.md index a49f0132303..c489f332460 100644 --- a/infra/charts/feast/README.md +++ b/infra/charts/feast/README.md @@ -85,7 +85,7 @@ The following table lists the configurable parameters of the Feast chart and the | `feast-core.prometheus-statsd-exporter.*` | Refer to this [link](charts/feast-core/charts/prometheus-statsd-exporter/values.yaml | | `feast-core.replicaCount` | No of pods to create | `1` | `feast-core.image.repository` | Repository for Feast Core Docker image | `gcr.io/kf-feast/feast-core` -| `feast-core.image.tag` | Tag for Feast Core Docker image | `0.4.6` +| `feast-core.image.tag` | Tag for Feast Core Docker image | `0.4.7` | `feast-core.image.pullPolicy` | Image pull policy for Feast Core Docker image | `IfNotPresent` | `feast-core.prometheus.enabled` | Add annotations to enable Prometheus scraping | `false` | `feast-core.application.yaml` | Configuration for Feast Core application | Refer to this [link](charts/feast-core/values.yaml) @@ -126,7 +126,7 @@ The following table lists the configurable parameters of the Feast chart and the | `feast-serving-online.core.enabled` | Flag for Feast Serving to use Feast Core in the same Helm release | `true` | `feast-serving-online.replicaCount` | No of pods to create | `1` | `feast-serving-online.image.repository` | Repository for Feast Serving Docker image | `gcr.io/kf-feast/feast-serving` -| `feast-serving-online.image.tag` | Tag for Feast Serving Docker image | `0.4.6` +| `feast-serving-online.image.tag` | Tag for Feast Serving Docker image | `0.4.7` | `feast-serving-online.image.pullPolicy` | Image pull policy for Feast Serving Docker image | `IfNotPresent` | `feast-serving-online.prometheus.enabled` | Add annotations to enable Prometheus scraping | `true` | `feast-serving-online.application.yaml` | Application configuration for Feast Serving | Refer to this [link](charts/feast-serving/values.yaml) @@ -168,7 +168,7 @@ The following table lists the configurable parameters of the Feast chart and the | `feast-serving-batch.core.enabled` | Flag for Feast Serving to use Feast Core in the same Helm release | `true` | `feast-serving-batch.replicaCount` | No of pods to create | `1` | `feast-serving-batch.image.repository` | Repository for Feast Serving Docker image | `gcr.io/kf-feast/feast-serving` -| `feast-serving-batch.image.tag` | Tag for Feast Serving Docker image | `0.4.6` +| `feast-serving-batch.image.tag` | Tag for Feast Serving Docker image | `0.4.7` | `feast-serving-batch.image.pullPolicy` | Image pull policy for Feast Serving Docker image | `IfNotPresent` | `feast-serving-batch.prometheus.enabled` | Add annotations to enable Prometheus scraping | `true` | `feast-serving-batch.application.yaml` | Application configuration for Feast Serving | Refer to this [link](charts/feast-serving/values.yaml) diff --git a/infra/charts/feast/charts/feast-core/Chart.yaml b/infra/charts/feast/charts/feast-core/Chart.yaml index 0f437d2b6c0..7587d795bf8 100644 --- a/infra/charts/feast/charts/feast-core/Chart.yaml +++ b/infra/charts/feast/charts/feast-core/Chart.yaml @@ -1,4 +1,4 @@ apiVersion: v1 description: A Helm chart for core component of Feast name: feast-core -version: 0.4.6 +version: 0.4.7 diff --git a/infra/charts/feast/charts/feast-serving/Chart.yaml b/infra/charts/feast/charts/feast-serving/Chart.yaml index d84d3377df2..d37877d29c2 100644 --- a/infra/charts/feast/charts/feast-serving/Chart.yaml +++ b/infra/charts/feast/charts/feast-serving/Chart.yaml @@ -1,4 +1,4 @@ apiVersion: v1 description: A Helm chart for serving component of Feast name: feast-serving -version: 0.4.6 +version: 0.4.7 diff --git a/infra/charts/feast/requirements.yaml b/infra/charts/feast/requirements.yaml index 3ed12f08871..ff9a0155ffa 100644 --- a/infra/charts/feast/requirements.yaml +++ b/infra/charts/feast/requirements.yaml @@ -1,12 +1,12 @@ dependencies: - name: feast-core - version: 0.4.6 + version: 0.4.7 condition: feast-core.enabled - name: feast-serving alias: feast-serving-batch - version: 0.4.6 + version: 0.4.7 condition: feast-serving-batch.enabled - name: feast-serving alias: feast-serving-online - version: 0.4.6 + version: 0.4.7 condition: feast-serving-online.enabled \ No newline at end of file diff --git a/infra/charts/feast/values.yaml b/infra/charts/feast/values.yaml index dd2174ae46d..c8afbf7109f 100644 --- a/infra/charts/feast/values.yaml +++ b/infra/charts/feast/values.yaml @@ -53,7 +53,7 @@ feast-core: # Specify which image tag to use. Keep this consistent for all components image: - tag: "0.4.5" + tag: "0.4.7" # jvmOptions are options that will be passed to the Java Virtual Machine (JVM) # running Feast Core. @@ -121,7 +121,7 @@ feast-serving-online: enabled: true # Specify what image tag to use. Keep this consistent for all components image: - tag: "0.4.5" + tag: "0.4.7" # redis.enabled specifies whether Redis should be installed as part of Feast Serving. # # If enabled is set to "false", Feast admin has to ensure there is an @@ -180,7 +180,7 @@ feast-serving-batch: enabled: true # Specify what image tag to use. Keep this consistent for all components image: - tag: "0.4.5" + tag: "0.4.7" # redis.enabled specifies whether Redis should be installed as part of Feast Serving. # # This is usually set to "false" for Feast Serving Batch because the default diff --git a/ingestion/pom.xml b/ingestion/pom.xml index 001da1a1453..ccc8ca04510 100644 --- a/ingestion/pom.xml +++ b/ingestion/pom.xml @@ -92,24 +92,6 @@ ${project.version} - - org.glassfish - javax.el - 3.0.0 - - - - javax.validation - validation-api - 2.0.1.Final - - - - org.hibernate.validator - hibernate-validator - 6.1.0.Final - - com.google.auto.value auto-value-annotations @@ -122,15 +104,6 @@ provided - - io.grpc - grpc-stub - - - - com.google.cloud - google-cloud-storage - com.google.cloud google-cloud-bigquery @@ -150,27 +123,6 @@ mockito-core - - com.fasterxml.jackson.core - jackson-annotations - - - com.fasterxml.jackson.core - jackson-core - - - com.fasterxml.jackson.core - jackson-databind - - - com.fasterxml.jackson.dataformat - jackson-dataformat-yaml - - - com.fasterxml.jackson.module - jackson-module-jsonSchema - - com.google.protobuf protobuf-java @@ -216,8 +168,8 @@ - redis.clients - jedis + io.lettuce + lettuce-core diff --git a/ingestion/src/main/java/feast/ingestion/transform/WriteToStore.java b/ingestion/src/main/java/feast/ingestion/transform/WriteToStore.java index b7901c2f90c..4e9082f5554 100644 --- a/ingestion/src/main/java/feast/ingestion/transform/WriteToStore.java +++ b/ingestion/src/main/java/feast/ingestion/transform/WriteToStore.java @@ -22,7 +22,6 @@ import feast.core.FeatureSetProto.FeatureSet; import feast.core.StoreProto.Store; import feast.core.StoreProto.Store.BigQueryConfig; -import feast.core.StoreProto.Store.RedisConfig; import feast.core.StoreProto.Store.StoreType; import feast.ingestion.options.ImportOptions; import feast.ingestion.utils.ResourceUtil; @@ -88,13 +87,12 @@ public PDone expand(PCollection input) { switch (storeType) { case REDIS: - RedisConfig redisConfig = getStore().getRedisConfig(); PCollection redisWriteResult = input .apply( "FeatureRowToRedisMutation", ParDo.of(new FeatureRowToRedisMutationDoFn(getFeatureSets()))) - .apply("WriteRedisMutationToRedis", RedisCustomIO.write(redisConfig)); + .apply("WriteRedisMutationToRedis", RedisCustomIO.write(getStore())); if (options.getDeadLetterTableSpec() != null) { redisWriteResult.apply( WriteFailedElementToBigQuery.newBuilder() diff --git a/ingestion/src/main/java/feast/ingestion/utils/StoreUtil.java b/ingestion/src/main/java/feast/ingestion/utils/StoreUtil.java index 7af98fb8f00..a02b8626945 100644 --- a/ingestion/src/main/java/feast/ingestion/utils/StoreUtil.java +++ b/ingestion/src/main/java/feast/ingestion/utils/StoreUtil.java @@ -43,14 +43,15 @@ import feast.core.StoreProto.Store.RedisConfig; import feast.core.StoreProto.Store.StoreType; import feast.types.ValueProto.ValueType.Enum; +import io.lettuce.core.RedisClient; +import io.lettuce.core.RedisConnectionException; +import io.lettuce.core.RedisURI; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; import org.apache.commons.lang3.tuple.Pair; import org.slf4j.Logger; -import redis.clients.jedis.JedisPool; -import redis.clients.jedis.exceptions.JedisConnectionException; // TODO: Create partitioned table by default @@ -239,15 +240,16 @@ public static void setupBigQuery( * @param redisConfig Plase refer to feast.core.Store proto */ public static void checkRedisConnection(RedisConfig redisConfig) { - JedisPool jedisPool = new JedisPool(redisConfig.getHost(), redisConfig.getPort()); + RedisClient redisClient = + RedisClient.create(RedisURI.create(redisConfig.getHost(), redisConfig.getPort())); try { - jedisPool.getResource(); - } catch (JedisConnectionException e) { + redisClient.connect(); + } catch (RedisConnectionException e) { throw new RuntimeException( String.format( "Failed to connect to Redis at host: '%s' port: '%d'. Please check that your Redis is running and accessible from Feast.", redisConfig.getHost(), redisConfig.getPort())); } - jedisPool.close(); + redisClient.shutdown(); } } diff --git a/ingestion/src/main/java/feast/retry/Retriable.java b/ingestion/src/main/java/feast/retry/Retriable.java index 0a788fcdd69..30676fe8208 100644 --- a/ingestion/src/main/java/feast/retry/Retriable.java +++ b/ingestion/src/main/java/feast/retry/Retriable.java @@ -17,7 +17,7 @@ package feast.retry; public interface Retriable { - void execute(); + void execute() throws Exception; Boolean isExceptionRetriable(Exception e); diff --git a/ingestion/src/main/java/feast/store/serving/redis/FeatureRowToRedisMutationDoFn.java b/ingestion/src/main/java/feast/store/serving/redis/FeatureRowToRedisMutationDoFn.java index 4b744d0fe6b..ca017c1f756 100644 --- a/ingestion/src/main/java/feast/store/serving/redis/FeatureRowToRedisMutationDoFn.java +++ b/ingestion/src/main/java/feast/store/serving/redis/FeatureRowToRedisMutationDoFn.java @@ -18,12 +18,15 @@ import feast.core.FeatureSetProto.EntitySpec; import feast.core.FeatureSetProto.FeatureSet; +import feast.core.FeatureSetProto.FeatureSetSpec; +import feast.core.FeatureSetProto.FeatureSpec; import feast.storage.RedisProto.RedisKey; import feast.storage.RedisProto.RedisKey.Builder; import feast.store.serving.redis.RedisCustomIO.Method; import feast.store.serving.redis.RedisCustomIO.RedisMutation; import feast.types.FeatureRowProto.FeatureRow; import feast.types.FieldProto.Field; +import feast.types.ValueProto; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -64,14 +67,45 @@ private RedisKey getKey(FeatureRow featureRow) { return redisKeyBuilder.build(); } + private byte[] getValue(FeatureRow featureRow) { + FeatureSetSpec spec = featureSets.get(featureRow.getFeatureSet()).getSpec(); + + List featureNames = + spec.getFeaturesList().stream().map(FeatureSpec::getName).collect(Collectors.toList()); + Map fieldValueOnlyMap = + featureRow.getFieldsList().stream() + .filter(field -> featureNames.contains(field.getName())) + .distinct() + .collect( + Collectors.toMap( + Field::getName, + field -> Field.newBuilder().setValue(field.getValue()).build())); + + List values = + featureNames.stream() + .sorted() + .map( + featureName -> + fieldValueOnlyMap.getOrDefault( + featureName, + Field.newBuilder().setValue(ValueProto.Value.getDefaultInstance()).build())) + .collect(Collectors.toList()); + + return FeatureRow.newBuilder() + .setEventTimestamp(featureRow.getEventTimestamp()) + .addAllFields(values) + .build() + .toByteArray(); + } + /** Output a redis mutation object for every feature in the feature row. */ @ProcessElement public void processElement(ProcessContext context) { FeatureRow featureRow = context.element(); try { - RedisKey key = getKey(featureRow); - RedisMutation redisMutation = - new RedisMutation(Method.SET, key.toByteArray(), featureRow.toByteArray(), null, null); + byte[] key = getKey(featureRow).toByteArray(); + byte[] value = getValue(featureRow); + RedisMutation redisMutation = new RedisMutation(Method.SET, key, value, null, null); context.output(redisMutation); } catch (Exception e) { log.error(e.getMessage(), e); diff --git a/ingestion/src/main/java/feast/store/serving/redis/RedisCustomIO.java b/ingestion/src/main/java/feast/store/serving/redis/RedisCustomIO.java index 8541baaffc3..633c2eb551d 100644 --- a/ingestion/src/main/java/feast/store/serving/redis/RedisCustomIO.java +++ b/ingestion/src/main/java/feast/store/serving/redis/RedisCustomIO.java @@ -18,11 +18,13 @@ import feast.core.StoreProto; import feast.ingestion.values.FailedElement; -import feast.retry.BackOffExecutor; import feast.retry.Retriable; +import io.lettuce.core.RedisConnectionException; import java.io.IOException; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; +import java.util.concurrent.ExecutionException; import org.apache.avro.reflect.Nullable; import org.apache.beam.sdk.coders.AvroCoder; import org.apache.beam.sdk.coders.DefaultCoder; @@ -32,14 +34,9 @@ import org.apache.beam.sdk.transforms.windowing.GlobalWindow; import org.apache.beam.sdk.values.PCollection; import org.apache.commons.lang3.exception.ExceptionUtils; -import org.joda.time.Duration; import org.joda.time.Instant; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import redis.clients.jedis.Jedis; -import redis.clients.jedis.Pipeline; -import redis.clients.jedis.Response; -import redis.clients.jedis.exceptions.JedisConnectionException; public class RedisCustomIO { @@ -50,8 +47,8 @@ public class RedisCustomIO { private RedisCustomIO() {} - public static Write write(StoreProto.Store.RedisConfig redisConfig) { - return new Write(redisConfig); + public static Write write(StoreProto.Store store) { + return new Write(store); } public enum Method { @@ -168,8 +165,8 @@ public static class Write private WriteDoFn dofn; - private Write(StoreProto.Store.RedisConfig redisConfig) { - this.dofn = new WriteDoFn(redisConfig); + private Write(StoreProto.Store store) { + this.dofn = new WriteDoFn(store); } public Write withBatchSize(int batchSize) { @@ -189,23 +186,14 @@ public PCollection expand(PCollection input) { public static class WriteDoFn extends DoFn { - private final String host; - private final int port; - private final BackOffExecutor backOffExecutor; private final List mutations = new ArrayList<>(); - - private Jedis jedis; - private Pipeline pipeline; private int batchSize = DEFAULT_BATCH_SIZE; private int timeout = DEFAULT_TIMEOUT; + private RedisIngestionClient redisIngestionClient; - WriteDoFn(StoreProto.Store.RedisConfig redisConfig) { - this.host = redisConfig.getHost(); - this.port = redisConfig.getPort(); - long backoffMs = - redisConfig.getInitialBackoffMs() > 0 ? redisConfig.getInitialBackoffMs() : 1; - this.backOffExecutor = - new BackOffExecutor(redisConfig.getMaxRetries(), Duration.millis(backoffMs)); + WriteDoFn(StoreProto.Store store) { + if (store.getType() == StoreProto.Store.StoreType.REDIS) + this.redisIngestionClient = new RedisStandaloneIngestionClient(store.getRedisConfig()); } public WriteDoFn withBatchSize(int batchSize) { @@ -224,47 +212,50 @@ public WriteDoFn withTimeout(int timeout) { @Setup public void setup() { - jedis = new Jedis(host, port, timeout); + this.redisIngestionClient.setup(); } @StartBundle public void startBundle() { + try { + redisIngestionClient.connect(); + } catch (RedisConnectionException e) { + log.error("Connection to redis cannot be established ", e); + } mutations.clear(); - pipeline = jedis.pipelined(); } private void executeBatch() throws Exception { - backOffExecutor.execute( - new Retriable() { - @Override - public void execute() { - mutations.forEach( - mutation -> { - writeRecord(mutation); - if (mutation.getExpiryMillis() != null && mutation.getExpiryMillis() > 0) { - pipeline.pexpire(mutation.getKey(), mutation.getExpiryMillis()); - } - }); - pipeline.sync(); - mutations.clear(); - } - - @Override - public Boolean isExceptionRetriable(Exception e) { - return e instanceof JedisConnectionException; - } - - @Override - public void cleanUpAfterFailure() { - try { - pipeline.close(); - } catch (IOException e) { - log.error(String.format("Error while closing pipeline: %s", e.getMessage())); - } - jedis = new Jedis(host, port, timeout); - pipeline = jedis.pipelined(); - } - }); + this.redisIngestionClient + .getBackOffExecutor() + .execute( + new Retriable() { + @Override + public void execute() throws ExecutionException, InterruptedException { + if (!redisIngestionClient.isConnected()) { + redisIngestionClient.connect(); + } + mutations.forEach( + mutation -> { + writeRecord(mutation); + if (mutation.getExpiryMillis() != null + && mutation.getExpiryMillis() > 0) { + redisIngestionClient.pexpire( + mutation.getKey(), mutation.getExpiryMillis()); + } + }); + redisIngestionClient.sync(); + mutations.clear(); + } + + @Override + public Boolean isExceptionRetriable(Exception e) { + return e instanceof RedisConnectionException; + } + + @Override + public void cleanUpAfterFailure() {} + }); } private FailedElement toFailedElement( @@ -272,7 +263,7 @@ private FailedElement toFailedElement( return FailedElement.newBuilder() .setJobName(jobName) .setTransformName("RedisCustomIO") - .setPayload(mutation.getValue().toString()) + .setPayload(Arrays.toString(mutation.getValue())) .setErrorMessage(exception.getMessage()) .setStackTrace(ExceptionUtils.getStackTrace(exception)) .build(); @@ -297,20 +288,26 @@ public void processElement(ProcessContext context) { } } - private Response writeRecord(RedisMutation mutation) { + private void writeRecord(RedisMutation mutation) { switch (mutation.getMethod()) { case APPEND: - return pipeline.append(mutation.getKey(), mutation.getValue()); + redisIngestionClient.append(mutation.getKey(), mutation.getValue()); + return; case SET: - return pipeline.set(mutation.getKey(), mutation.getValue()); + redisIngestionClient.set(mutation.getKey(), mutation.getValue()); + return; case LPUSH: - return pipeline.lpush(mutation.getKey(), mutation.getValue()); + redisIngestionClient.lpush(mutation.getKey(), mutation.getValue()); + return; case RPUSH: - return pipeline.rpush(mutation.getKey(), mutation.getValue()); + redisIngestionClient.rpush(mutation.getKey(), mutation.getValue()); + return; case SADD: - return pipeline.sadd(mutation.getKey(), mutation.getValue()); + redisIngestionClient.sadd(mutation.getKey(), mutation.getValue()); + return; case ZADD: - return pipeline.zadd(mutation.getKey(), mutation.getScore(), mutation.getValue()); + redisIngestionClient.zadd(mutation.getKey(), mutation.getScore(), mutation.getValue()); + return; default: throw new UnsupportedOperationException( String.format("Not implemented writing records for %s", mutation.getMethod())); @@ -337,7 +334,7 @@ public void finishBundle(FinishBundleContext context) @Teardown public void teardown() { - jedis.close(); + redisIngestionClient.shutdown(); } } } diff --git a/ingestion/src/main/java/feast/store/serving/redis/RedisIngestionClient.java b/ingestion/src/main/java/feast/store/serving/redis/RedisIngestionClient.java new file mode 100644 index 00000000000..d51eead53fb --- /dev/null +++ b/ingestion/src/main/java/feast/store/serving/redis/RedisIngestionClient.java @@ -0,0 +1,49 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2020 The Feast Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package feast.store.serving.redis; + +import feast.retry.BackOffExecutor; +import java.io.Serializable; + +public interface RedisIngestionClient extends Serializable { + + void setup(); + + BackOffExecutor getBackOffExecutor(); + + void shutdown(); + + void connect(); + + boolean isConnected(); + + void sync(); + + void pexpire(byte[] key, Long expiryMillis); + + void append(byte[] key, byte[] value); + + void set(byte[] key, byte[] value); + + void lpush(byte[] key, byte[] value); + + void rpush(byte[] key, byte[] value); + + void sadd(byte[] key, byte[] value); + + void zadd(byte[] key, Long score, byte[] value); +} diff --git a/ingestion/src/main/java/feast/store/serving/redis/RedisStandaloneIngestionClient.java b/ingestion/src/main/java/feast/store/serving/redis/RedisStandaloneIngestionClient.java new file mode 100644 index 00000000000..d95ebbbf64a --- /dev/null +++ b/ingestion/src/main/java/feast/store/serving/redis/RedisStandaloneIngestionClient.java @@ -0,0 +1,122 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2020 The Feast Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package feast.store.serving.redis; + +import com.google.common.collect.Lists; +import feast.core.StoreProto; +import feast.retry.BackOffExecutor; +import io.lettuce.core.*; +import io.lettuce.core.api.StatefulRedisConnection; +import io.lettuce.core.api.async.RedisAsyncCommands; +import io.lettuce.core.codec.ByteArrayCodec; +import java.util.List; +import java.util.concurrent.TimeUnit; +import org.joda.time.Duration; + +public class RedisStandaloneIngestionClient implements RedisIngestionClient { + private final String host; + private final int port; + private final BackOffExecutor backOffExecutor; + private RedisClient redisclient; + private static final int DEFAULT_TIMEOUT = 2000; + private StatefulRedisConnection connection; + private RedisAsyncCommands commands; + private List futures = Lists.newArrayList(); + + public RedisStandaloneIngestionClient(StoreProto.Store.RedisConfig redisConfig) { + this.host = redisConfig.getHost(); + this.port = redisConfig.getPort(); + long backoffMs = redisConfig.getInitialBackoffMs() > 0 ? redisConfig.getInitialBackoffMs() : 1; + this.backOffExecutor = + new BackOffExecutor(redisConfig.getMaxRetries(), Duration.millis(backoffMs)); + } + + @Override + public void setup() { + this.redisclient = + RedisClient.create(new RedisURI(host, port, java.time.Duration.ofMillis(DEFAULT_TIMEOUT))); + } + + @Override + public BackOffExecutor getBackOffExecutor() { + return this.backOffExecutor; + } + + @Override + public void shutdown() { + this.redisclient.shutdown(); + } + + @Override + public void connect() { + if (!isConnected()) { + this.connection = this.redisclient.connect(new ByteArrayCodec()); + this.commands = connection.async(); + } + } + + @Override + public boolean isConnected() { + return connection != null; + } + + @Override + public void sync() { + // Wait for some time for futures to complete + // TODO: should this be configurable? + try { + LettuceFutures.awaitAll(60, TimeUnit.SECONDS, futures.toArray(new RedisFuture[0])); + } finally { + futures.clear(); + } + } + + @Override + public void pexpire(byte[] key, Long expiryMillis) { + commands.pexpire(key, expiryMillis); + } + + @Override + public void append(byte[] key, byte[] value) { + futures.add(commands.append(key, value)); + } + + @Override + public void set(byte[] key, byte[] value) { + futures.add(commands.set(key, value)); + } + + @Override + public void lpush(byte[] key, byte[] value) { + futures.add(commands.lpush(key, value)); + } + + @Override + public void rpush(byte[] key, byte[] value) { + futures.add(commands.rpush(key, value)); + } + + @Override + public void sadd(byte[] key, byte[] value) { + futures.add(commands.sadd(key, value)); + } + + @Override + public void zadd(byte[] key, Long score, byte[] value) { + futures.add(commands.zadd(key, score, value)); + } +} diff --git a/ingestion/src/test/java/feast/ingestion/ImportJobTest.java b/ingestion/src/test/java/feast/ingestion/ImportJobTest.java index 1148fa40422..0b000df0f59 100644 --- a/ingestion/src/test/java/feast/ingestion/ImportJobTest.java +++ b/ingestion/src/test/java/feast/ingestion/ImportJobTest.java @@ -37,7 +37,13 @@ import feast.test.TestUtil.LocalKafka; import feast.test.TestUtil.LocalRedis; import feast.types.FeatureRowProto.FeatureRow; +import feast.types.FieldProto; import feast.types.ValueProto.ValueType.Enum; +import io.lettuce.core.RedisClient; +import io.lettuce.core.RedisURI; +import io.lettuce.core.api.StatefulRedisConnection; +import io.lettuce.core.api.sync.RedisCommands; +import io.lettuce.core.codec.ByteArrayCodec; import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.ArrayList; @@ -45,6 +51,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.stream.Collectors; import java.util.stream.IntStream; import org.apache.beam.sdk.PipelineResult; import org.apache.beam.sdk.PipelineResult.State; @@ -57,7 +64,6 @@ import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import redis.clients.jedis.Jedis; public class ImportJobTest { @@ -185,6 +191,23 @@ public void runPipeline_ShouldWriteToRedisCorrectlyGivenValidSpecAndFeatureRow() FeatureRow randomRow = TestUtil.createRandomFeatureRow(featureSet); RedisKey redisKey = TestUtil.createRedisKey(featureSet, randomRow); input.add(randomRow); + List fields = + randomRow.getFieldsList().stream() + .filter( + field -> + spec.getFeaturesList().stream() + .map(FeatureSpec::getName) + .collect(Collectors.toList()) + .contains(field.getName())) + .map(field -> field.toBuilder().clearName().build()) + .collect(Collectors.toList()); + randomRow = + randomRow + .toBuilder() + .clearFields() + .addAllFields(fields) + .clearFeatureSet() + .build(); expected.put(redisKey, randomRow); }); @@ -206,21 +229,24 @@ public void runPipeline_ShouldWriteToRedisCorrectlyGivenValidSpecAndFeatureRow() Duration.standardSeconds(IMPORT_JOB_CHECK_INTERVAL_DURATION_SEC)); LOGGER.info("Validating the actual values written to Redis ..."); - Jedis jedis = new Jedis(REDIS_HOST, REDIS_PORT); + RedisClient redisClient = + RedisClient.create(new RedisURI(REDIS_HOST, REDIS_PORT, java.time.Duration.ofMillis(2000))); + StatefulRedisConnection connection = redisClient.connect(new ByteArrayCodec()); + RedisCommands sync = connection.sync(); expected.forEach( (key, expectedValue) -> { // Ensure ingested key exists. - byte[] actualByteValue = jedis.get(key.toByteArray()); + byte[] actualByteValue = sync.get(key.toByteArray()); if (actualByteValue == null) { LOGGER.error("Key not found in Redis: " + key); LOGGER.info("Redis INFO:"); - LOGGER.info(jedis.info()); - String randomKey = jedis.randomKey(); + LOGGER.info(sync.info()); + byte[] randomKey = sync.randomkey(); if (randomKey != null) { LOGGER.info("Sample random key, value (for debugging purpose):"); LOGGER.info("Key: " + randomKey); - LOGGER.info("Value: " + jedis.get(randomKey)); + LOGGER.info("Value: " + sync.get(randomKey)); } Assert.fail("Missing key in Redis."); } @@ -239,5 +265,6 @@ public void runPipeline_ShouldWriteToRedisCorrectlyGivenValidSpecAndFeatureRow() // Ensure the retrieved FeatureRow is equal to the ingested FeatureRow. Assert.assertEquals(expectedValue, actualValue); }); + redisClient.shutdown(); } } diff --git a/ingestion/src/test/java/feast/store/serving/redis/FeatureRowToRedisMutationDoFnTest.java b/ingestion/src/test/java/feast/store/serving/redis/FeatureRowToRedisMutationDoFnTest.java index 92bb6e41c38..86b4feae05f 100644 --- a/ingestion/src/test/java/feast/store/serving/redis/FeatureRowToRedisMutationDoFnTest.java +++ b/ingestion/src/test/java/feast/store/serving/redis/FeatureRowToRedisMutationDoFnTest.java @@ -29,10 +29,7 @@ import feast.types.FieldProto.Field; import feast.types.ValueProto.Value; import feast.types.ValueProto.ValueType.Enum; -import java.util.Arrays; -import java.util.Collections; -import java.util.HashMap; -import java.util.Map; +import java.util.*; import org.apache.beam.sdk.extensions.protobuf.ProtoCoder; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; @@ -96,6 +93,14 @@ public void shouldConvertRowWithDuplicateEntitiesToValidKey() { Field.newBuilder() .setName("entity_id_secondary") .setValue(Value.newBuilder().setStringVal("a"))) + .addFields( + Field.newBuilder() + .setName("feature_1") + .setValue(Value.newBuilder().setStringVal("strValue1"))) + .addFields( + Field.newBuilder() + .setName("feature_2") + .setValue(Value.newBuilder().setInt64Val(1001))) .build(); PCollection output = @@ -116,6 +121,13 @@ public void shouldConvertRowWithDuplicateEntitiesToValidKey() { .setValue(Value.newBuilder().setStringVal("a"))) .build(); + FeatureRow expectedValue = + FeatureRow.newBuilder() + .setEventTimestamp(Timestamp.newBuilder().setSeconds(10)) + .addFields(Field.newBuilder().setValue(Value.newBuilder().setStringVal("strValue1"))) + .addFields(Field.newBuilder().setValue(Value.newBuilder().setInt64Val(1001))) + .build(); + PAssert.that(output) .satisfies( (SerializableFunction, Void>) @@ -123,7 +135,7 @@ public void shouldConvertRowWithDuplicateEntitiesToValidKey() { input.forEach( rm -> { assert (Arrays.equals(rm.getKey(), expectedKey.toByteArray())); - assert (Arrays.equals(rm.getValue(), offendingRow.toByteArray())); + assert (Arrays.equals(rm.getValue(), expectedValue.toByteArray())); }); return null; }); @@ -131,7 +143,7 @@ public void shouldConvertRowWithDuplicateEntitiesToValidKey() { } @Test - public void shouldConvertRowWithOutOfOrderEntitiesToValidKey() { + public void shouldConvertRowWithOutOfOrderFieldsToValidKey() { Map featureSets = new HashMap<>(); featureSets.put("feature_set", fs); @@ -147,6 +159,14 @@ public void shouldConvertRowWithOutOfOrderEntitiesToValidKey() { Field.newBuilder() .setName("entity_id_primary") .setValue(Value.newBuilder().setInt32Val(1))) + .addFields( + Field.newBuilder() + .setName("feature_2") + .setValue(Value.newBuilder().setInt64Val(1001))) + .addFields( + Field.newBuilder() + .setName("feature_1") + .setValue(Value.newBuilder().setStringVal("strValue1"))) .build(); PCollection output = @@ -167,6 +187,148 @@ public void shouldConvertRowWithOutOfOrderEntitiesToValidKey() { .setValue(Value.newBuilder().setStringVal("a"))) .build(); + List expectedFields = + Arrays.asList( + Field.newBuilder().setValue(Value.newBuilder().setStringVal("strValue1")).build(), + Field.newBuilder().setValue(Value.newBuilder().setInt64Val(1001)).build()); + FeatureRow expectedValue = + FeatureRow.newBuilder() + .setEventTimestamp(Timestamp.newBuilder().setSeconds(10)) + .addAllFields(expectedFields) + .build(); + + PAssert.that(output) + .satisfies( + (SerializableFunction, Void>) + input -> { + input.forEach( + rm -> { + assert (Arrays.equals(rm.getKey(), expectedKey.toByteArray())); + assert (Arrays.equals(rm.getValue(), expectedValue.toByteArray())); + }); + return null; + }); + p.run(); + } + + @Test + public void shouldMergeDuplicateFeatureFields() { + Map featureSets = new HashMap<>(); + featureSets.put("feature_set", fs); + + FeatureRow featureRowWithDuplicatedFeatureFields = + FeatureRow.newBuilder() + .setFeatureSet("feature_set") + .setEventTimestamp(Timestamp.newBuilder().setSeconds(10)) + .addFields( + Field.newBuilder() + .setName("entity_id_primary") + .setValue(Value.newBuilder().setInt32Val(1))) + .addFields( + Field.newBuilder() + .setName("entity_id_secondary") + .setValue(Value.newBuilder().setStringVal("a"))) + .addFields( + Field.newBuilder() + .setName("feature_1") + .setValue(Value.newBuilder().setStringVal("strValue1"))) + .addFields( + Field.newBuilder() + .setName("feature_1") + .setValue(Value.newBuilder().setStringVal("strValue1"))) + .addFields( + Field.newBuilder() + .setName("feature_2") + .setValue(Value.newBuilder().setInt64Val(1001))) + .build(); + + PCollection output = + p.apply(Create.of(Collections.singletonList(featureRowWithDuplicatedFeatureFields))) + .setCoder(ProtoCoder.of(FeatureRow.class)) + .apply(ParDo.of(new FeatureRowToRedisMutationDoFn(featureSets))); + + RedisKey expectedKey = + RedisKey.newBuilder() + .setFeatureSet("feature_set") + .addEntities( + Field.newBuilder() + .setName("entity_id_primary") + .setValue(Value.newBuilder().setInt32Val(1))) + .addEntities( + Field.newBuilder() + .setName("entity_id_secondary") + .setValue(Value.newBuilder().setStringVal("a"))) + .build(); + + FeatureRow expectedValue = + FeatureRow.newBuilder() + .setEventTimestamp(Timestamp.newBuilder().setSeconds(10)) + .addFields(Field.newBuilder().setValue(Value.newBuilder().setStringVal("strValue1"))) + .addFields(Field.newBuilder().setValue(Value.newBuilder().setInt64Val(1001))) + .build(); + + PAssert.that(output) + .satisfies( + (SerializableFunction, Void>) + input -> { + input.forEach( + rm -> { + assert (Arrays.equals(rm.getKey(), expectedKey.toByteArray())); + assert (Arrays.equals(rm.getValue(), expectedValue.toByteArray())); + }); + return null; + }); + p.run(); + } + + @Test + public void shouldPopulateMissingFeatureValuesWithDefaultInstance() { + Map featureSets = new HashMap<>(); + featureSets.put("feature_set", fs); + + FeatureRow featureRowWithDuplicatedFeatureFields = + FeatureRow.newBuilder() + .setFeatureSet("feature_set") + .setEventTimestamp(Timestamp.newBuilder().setSeconds(10)) + .addFields( + Field.newBuilder() + .setName("entity_id_primary") + .setValue(Value.newBuilder().setInt32Val(1))) + .addFields( + Field.newBuilder() + .setName("entity_id_secondary") + .setValue(Value.newBuilder().setStringVal("a"))) + .addFields( + Field.newBuilder() + .setName("feature_1") + .setValue(Value.newBuilder().setStringVal("strValue1"))) + .build(); + + PCollection output = + p.apply(Create.of(Collections.singletonList(featureRowWithDuplicatedFeatureFields))) + .setCoder(ProtoCoder.of(FeatureRow.class)) + .apply(ParDo.of(new FeatureRowToRedisMutationDoFn(featureSets))); + + RedisKey expectedKey = + RedisKey.newBuilder() + .setFeatureSet("feature_set") + .addEntities( + Field.newBuilder() + .setName("entity_id_primary") + .setValue(Value.newBuilder().setInt32Val(1))) + .addEntities( + Field.newBuilder() + .setName("entity_id_secondary") + .setValue(Value.newBuilder().setStringVal("a"))) + .build(); + + FeatureRow expectedValue = + FeatureRow.newBuilder() + .setEventTimestamp(Timestamp.newBuilder().setSeconds(10)) + .addFields(Field.newBuilder().setValue(Value.newBuilder().setStringVal("strValue1"))) + .addFields(Field.newBuilder().setValue(Value.getDefaultInstance())) + .build(); + PAssert.that(output) .satisfies( (SerializableFunction, Void>) @@ -174,7 +336,7 @@ public void shouldConvertRowWithOutOfOrderEntitiesToValidKey() { input.forEach( rm -> { assert (Arrays.equals(rm.getKey(), expectedKey.toByteArray())); - assert (Arrays.equals(rm.getValue(), offendingRow.toByteArray())); + assert (Arrays.equals(rm.getValue(), expectedValue.toByteArray())); }); return null; }); diff --git a/ingestion/src/test/java/feast/store/serving/redis/RedisCustomIOTest.java b/ingestion/src/test/java/feast/store/serving/redis/RedisCustomIOTest.java index fc17f6207f6..75663d24a6a 100644 --- a/ingestion/src/test/java/feast/store/serving/redis/RedisCustomIOTest.java +++ b/ingestion/src/test/java/feast/store/serving/redis/RedisCustomIOTest.java @@ -26,6 +26,11 @@ import feast.store.serving.redis.RedisCustomIO.RedisMutation; import feast.types.FeatureRowProto.FeatureRow; import feast.types.ValueProto.ValueType.Enum; +import io.lettuce.core.RedisClient; +import io.lettuce.core.RedisURI; +import io.lettuce.core.api.StatefulRedisConnection; +import io.lettuce.core.api.sync.RedisStringCommands; +import io.lettuce.core.codec.ByteArrayCodec; import java.io.IOException; import java.util.HashMap; import java.util.LinkedHashMap; @@ -43,7 +48,6 @@ import org.junit.Before; import org.junit.Rule; import org.junit.Test; -import redis.clients.jedis.Jedis; import redis.embedded.Redis; import redis.embedded.RedisServer; @@ -53,17 +57,22 @@ public class RedisCustomIOTest { private static String REDIS_HOST = "localhost"; private static int REDIS_PORT = 51234; private Redis redis; - private Jedis jedis; + private RedisClient redisClient; + private RedisStringCommands sync; @Before public void setUp() throws IOException { redis = new RedisServer(REDIS_PORT); redis.start(); - jedis = new Jedis(REDIS_HOST, REDIS_PORT); + redisClient = + RedisClient.create(new RedisURI(REDIS_HOST, REDIS_PORT, java.time.Duration.ofMillis(2000))); + StatefulRedisConnection connection = redisClient.connect(new ByteArrayCodec()); + sync = connection.sync(); } @After public void teardown() { + redisClient.shutdown(); redis.stop(); } @@ -105,12 +114,17 @@ public void shouldWriteToRedis() { null)) .collect(Collectors.toList()); - p.apply(Create.of(featureRowWrites)).apply(RedisCustomIO.write(redisConfig)); + StoreProto.Store store = + StoreProto.Store.newBuilder() + .setRedisConfig(redisConfig) + .setType(StoreProto.Store.StoreType.REDIS) + .build(); + p.apply(Create.of(featureRowWrites)).apply(RedisCustomIO.write(store)); p.run(); kvs.forEach( (key, value) -> { - byte[] actual = jedis.get(key.toByteArray()); + byte[] actual = sync.get(key.toByteArray()); assertThat(actual, equalTo(value.toByteArray())); }); } @@ -148,9 +162,14 @@ public void shouldRetryFailConnection() throws InterruptedException { null)) .collect(Collectors.toList()); + StoreProto.Store store = + StoreProto.Store.newBuilder() + .setRedisConfig(redisConfig) + .setType(StoreProto.Store.StoreType.REDIS) + .build(); PCollection failedElementCount = p.apply(Create.of(featureRowWrites)) - .apply(RedisCustomIO.write(redisConfig)) + .apply(RedisCustomIO.write(store)) .apply(Count.globally()); redis.stop(); @@ -169,7 +188,7 @@ public void shouldRetryFailConnection() throws InterruptedException { kvs.forEach( (key, value) -> { - byte[] actual = jedis.get(key.toByteArray()); + byte[] actual = sync.get(key.toByteArray()); assertThat(actual, equalTo(value.toByteArray())); }); } @@ -202,9 +221,14 @@ public void shouldProduceFailedElementIfRetryExceeded() { null)) .collect(Collectors.toList()); + StoreProto.Store store = + StoreProto.Store.newBuilder() + .setRedisConfig(redisConfig) + .setType(StoreProto.Store.StoreType.REDIS) + .build(); PCollection failedElementCount = p.apply(Create.of(featureRowWrites)) - .apply(RedisCustomIO.write(redisConfig)) + .apply(RedisCustomIO.write(store)) .apply(Count.globally()); redis.stop(); diff --git a/pom.xml b/pom.xml index b8310aca91d..fb50e531d7f 100644 --- a/pom.xml +++ b/pom.xml @@ -36,7 +36,7 @@ - 0.4.6-SNAPSHOT + 0.4.7-SNAPSHOT https://github.com/gojek/feast UTF-8 @@ -143,6 +143,11 @@ + + io.grpc + grpc-core + ${grpcVersion} + io.grpc grpc-netty @@ -278,6 +283,11 @@ log4j-jul ${log4jVersion} + + org.apache.logging.log4j + log4j-web + ${log4jVersion} + org.apache.logging.log4j log4j-slf4j-impl @@ -546,6 +556,20 @@ docker-maven-plugin 0.20.1 + + org.apache.maven.plugins + maven-dependency-plugin + 3.1.1 + + + + org.apache.maven.shared + maven-dependency-analyzer + 1.11.1 + + + org.apache.maven.plugins maven-javadoc-plugin diff --git a/sdk/python/setup.py b/sdk/python/setup.py index 3fc77540c02..9d8a3786505 100644 --- a/sdk/python/setup.py +++ b/sdk/python/setup.py @@ -37,9 +37,7 @@ "pandavro==1.5.*", "protobuf>=3.10", "PyYAML==5.1.*", - # fastavro 0.22.10 and newer will throw this error for e2e batch test: - # TypeError: Timestamp subtraction must have the same timezones or no timezones - "fastavro==0.22.9", + "fastavro>=0.22.11,<0.23", "kafka-python==1.*", "tabulate==0.8.*", "toml==0.10.*", diff --git a/serving/pom.xml b/serving/pom.xml index be573be45c5..4cc02dc4510 100644 --- a/serving/pom.xml +++ b/serving/pom.xml @@ -98,6 +98,10 @@ org.springframework.boot spring-boot-starter-log4j2 + + org.apache.logging.log4j + log4j-web + org.springframework.boot @@ -138,11 +142,11 @@ 3.1.0 - - redis.clients - jedis + io.lettuce + lettuce-core + com.google.guava diff --git a/serving/src/main/java/feast/serving/configuration/JobServiceConfig.java b/serving/src/main/java/feast/serving/configuration/JobServiceConfig.java index 4c6b652c46e..fa94dab8329 100644 --- a/serving/src/main/java/feast/serving/configuration/JobServiceConfig.java +++ b/serving/src/main/java/feast/serving/configuration/JobServiceConfig.java @@ -22,42 +22,24 @@ import feast.serving.service.NoopJobService; import feast.serving.service.RedisBackedJobService; import feast.serving.specs.CachedSpecService; -import java.util.Map; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import redis.clients.jedis.JedisPool; -import redis.clients.jedis.JedisPoolConfig; @Configuration public class JobServiceConfig { - public static final String DEFAULT_REDIS_MAX_CONN = "8"; - public static final String DEFAULT_REDIS_MAX_IDLE = "8"; - public static final String DEFAULT_REDIS_MAX_WAIT_MILLIS = "50"; - @Bean - public JobService jobService(FeastProperties feastProperties, CachedSpecService specService) { + public JobService jobService( + FeastProperties feastProperties, + CachedSpecService specService, + StoreConfiguration storeConfiguration) { if (!specService.getStore().getType().equals(StoreType.BIGQUERY)) { return new NoopJobService(); } StoreType storeType = StoreType.valueOf(feastProperties.getJobs().getStoreType()); - Map storeOptions = feastProperties.getJobs().getStoreOptions(); switch (storeType) { case REDIS: - JedisPoolConfig jedisPoolConfig = new JedisPoolConfig(); - jedisPoolConfig.setMaxTotal( - Integer.parseInt(storeOptions.getOrDefault("max-conn", DEFAULT_REDIS_MAX_CONN))); - jedisPoolConfig.setMaxIdle( - Integer.parseInt(storeOptions.getOrDefault("max-idle", DEFAULT_REDIS_MAX_IDLE))); - jedisPoolConfig.setMaxWaitMillis( - Integer.parseInt( - storeOptions.getOrDefault("max-wait-millis", DEFAULT_REDIS_MAX_WAIT_MILLIS))); - JedisPool jedisPool = - new JedisPool( - jedisPoolConfig, - storeOptions.get("host"), - Integer.parseInt(storeOptions.get("port"))); - return new RedisBackedJobService(jedisPool); + return new RedisBackedJobService(storeConfiguration.getJobStoreRedisConnection()); case INVALID: case BIGQUERY: case CASSANDRA: diff --git a/serving/src/main/java/feast/serving/configuration/ServingServiceConfig.java b/serving/src/main/java/feast/serving/configuration/ServingServiceConfig.java index 3cc115978a3..d0ea058baf4 100644 --- a/serving/src/main/java/feast/serving/configuration/ServingServiceConfig.java +++ b/serving/src/main/java/feast/serving/configuration/ServingServiceConfig.java @@ -36,8 +36,6 @@ import org.slf4j.Logger; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import redis.clients.jedis.JedisPool; -import redis.clients.jedis.JedisPoolConfig; @Configuration public class ServingServiceConfig { @@ -74,19 +72,16 @@ public ServingService servingService( FeastProperties feastProperties, CachedSpecService specService, JobService jobService, - Tracer tracer) { + Tracer tracer, + StoreConfiguration storeConfiguration) { ServingService servingService = null; Store store = specService.getStore(); switch (store.getType()) { case REDIS: - RedisConfig redisConfig = store.getRedisConfig(); - JedisPoolConfig poolConfig = new JedisPoolConfig(); - poolConfig.setMaxTotal(feastProperties.getStore().getRedisPoolMaxSize()); - poolConfig.setMaxIdle(feastProperties.getStore().getRedisPoolMaxIdle()); - JedisPool jedisPool = - new JedisPool(poolConfig, redisConfig.getHost(), redisConfig.getPort()); - servingService = new RedisServingService(jedisPool, specService, tracer); + servingService = + new RedisServingService( + storeConfiguration.getServingRedisConnection(), specService, tracer); break; case BIGQUERY: BigQueryConfig bqConfig = store.getBigqueryConfig(); diff --git a/serving/src/main/java/feast/serving/configuration/SpecServiceConfig.java b/serving/src/main/java/feast/serving/configuration/SpecServiceConfig.java index 0b3a2938b8e..26ebfa956ca 100644 --- a/serving/src/main/java/feast/serving/configuration/SpecServiceConfig.java +++ b/serving/src/main/java/feast/serving/configuration/SpecServiceConfig.java @@ -59,7 +59,6 @@ public ScheduledExecutorService cachedSpecServiceScheduledExecutorService( @Bean public CachedSpecService specService(FeastProperties feastProperties) { - CoreSpecService coreService = new CoreSpecService(feastCoreHost, feastCorePort); Path path = Paths.get(feastProperties.getStore().getConfigPath()); CachedSpecService cachedSpecStorage = new CachedSpecService(coreService, path); diff --git a/serving/src/main/java/feast/serving/configuration/StoreConfiguration.java b/serving/src/main/java/feast/serving/configuration/StoreConfiguration.java new file mode 100644 index 00000000000..84dc7b7f8d4 --- /dev/null +++ b/serving/src/main/java/feast/serving/configuration/StoreConfiguration.java @@ -0,0 +1,47 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2020 The Feast Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package feast.serving.configuration; + +import io.lettuce.core.api.StatefulRedisConnection; +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Configuration; + +@Configuration +public class StoreConfiguration { + + // We can define other store specific beans here + // These beans can be autowired or can be created in this class. + private final StatefulRedisConnection servingRedisConnection; + private final StatefulRedisConnection jobStoreRedisConnection; + + @Autowired + public StoreConfiguration( + ObjectProvider> servingRedisConnection, + ObjectProvider> jobStoreRedisConnection) { + this.servingRedisConnection = servingRedisConnection.getIfAvailable(); + this.jobStoreRedisConnection = jobStoreRedisConnection.getIfAvailable(); + } + + public StatefulRedisConnection getServingRedisConnection() { + return servingRedisConnection; + } + + public StatefulRedisConnection getJobStoreRedisConnection() { + return jobStoreRedisConnection; + } +} diff --git a/serving/src/main/java/feast/serving/configuration/redis/JobStoreRedisConfig.java b/serving/src/main/java/feast/serving/configuration/redis/JobStoreRedisConfig.java new file mode 100644 index 00000000000..77d9262bcb3 --- /dev/null +++ b/serving/src/main/java/feast/serving/configuration/redis/JobStoreRedisConfig.java @@ -0,0 +1,68 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2020 The Feast Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package feast.serving.configuration.redis; + +import com.google.common.base.Enums; +import feast.core.StoreProto; +import feast.serving.FeastProperties; +import io.lettuce.core.RedisClient; +import io.lettuce.core.RedisURI; +import io.lettuce.core.api.StatefulRedisConnection; +import io.lettuce.core.codec.ByteArrayCodec; +import io.lettuce.core.resource.ClientResources; +import io.lettuce.core.resource.DefaultClientResources; +import java.util.Map; +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +@Configuration +public class JobStoreRedisConfig { + + @Bean(destroyMethod = "shutdown") + ClientResources jobStoreClientResources() { + return DefaultClientResources.create(); + } + + @Bean(destroyMethod = "shutdown") + RedisClient jobStoreRedisClient( + ClientResources jobStoreClientResources, FeastProperties feastProperties) { + StoreProto.Store.StoreType storeType = + Enums.getIfPresent( + StoreProto.Store.StoreType.class, feastProperties.getJobs().getStoreType()) + .orNull(); + if (storeType != StoreProto.Store.StoreType.REDIS) return null; + Map jobStoreConf = feastProperties.getJobs().getStoreOptions(); + // If job conf is empty throw StoreException + if (jobStoreConf == null + || jobStoreConf.get("host") == null + || jobStoreConf.get("host").isEmpty() + || jobStoreConf.get("port") == null + || jobStoreConf.get("port").isEmpty()) + throw new IllegalArgumentException("Store Configuration is not set"); + RedisURI uri = + RedisURI.create(jobStoreConf.get("host"), Integer.parseInt(jobStoreConf.get("port"))); + return RedisClient.create(jobStoreClientResources, uri); + } + + @Bean(destroyMethod = "close") + StatefulRedisConnection jobStoreRedisConnection( + ObjectProvider jobStoreRedisClient) { + if (jobStoreRedisClient.getIfAvailable() == null) return null; + return jobStoreRedisClient.getIfAvailable().connect(new ByteArrayCodec()); + } +} diff --git a/serving/src/main/java/feast/serving/configuration/redis/ServingStoreRedisConfig.java b/serving/src/main/java/feast/serving/configuration/redis/ServingStoreRedisConfig.java new file mode 100644 index 00000000000..17a50eef6d6 --- /dev/null +++ b/serving/src/main/java/feast/serving/configuration/redis/ServingStoreRedisConfig.java @@ -0,0 +1,62 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2020 The Feast Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package feast.serving.configuration.redis; + +import feast.core.StoreProto; +import feast.serving.specs.CachedSpecService; +import io.lettuce.core.RedisClient; +import io.lettuce.core.RedisURI; +import io.lettuce.core.api.StatefulRedisConnection; +import io.lettuce.core.codec.ByteArrayCodec; +import io.lettuce.core.resource.ClientResources; +import io.lettuce.core.resource.DefaultClientResources; +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.context.annotation.*; + +@Configuration +public class ServingStoreRedisConfig { + + @Bean + StoreProto.Store.RedisConfig servingStoreRedisConf(CachedSpecService specService) { + if (specService.getStore().getType() != StoreProto.Store.StoreType.REDIS) return null; + return specService.getStore().getRedisConfig(); + } + + @Bean(destroyMethod = "shutdown") + ClientResources servingClientResources() { + return DefaultClientResources.create(); + } + + @Bean(destroyMethod = "shutdown") + RedisClient servingRedisClient( + ClientResources servingClientResources, + ObjectProvider servingStoreRedisConf) { + if (servingStoreRedisConf.getIfAvailable() == null) return null; + RedisURI redisURI = + RedisURI.create( + servingStoreRedisConf.getIfAvailable().getHost(), + servingStoreRedisConf.getIfAvailable().getPort()); + return RedisClient.create(servingClientResources, redisURI); + } + + @Bean(destroyMethod = "close") + StatefulRedisConnection servingRedisConnection( + ObjectProvider servingRedisClient) { + if (servingRedisClient.getIfAvailable() == null) return null; + return servingRedisClient.getIfAvailable().connect(new ByteArrayCodec()); + } +} diff --git a/serving/src/main/java/feast/serving/encoding/FeatureRowDecoder.java b/serving/src/main/java/feast/serving/encoding/FeatureRowDecoder.java new file mode 100644 index 00000000000..e70695d8c64 --- /dev/null +++ b/serving/src/main/java/feast/serving/encoding/FeatureRowDecoder.java @@ -0,0 +1,95 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2020 The Feast Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package feast.serving.encoding; + +import feast.core.FeatureSetProto.FeatureSetSpec; +import feast.core.FeatureSetProto.FeatureSpec; +import feast.types.FeatureRowProto.FeatureRow; +import feast.types.FieldProto.Field; +import java.util.Comparator; +import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +public class FeatureRowDecoder { + + private final String featureSetRef; + private final FeatureSetSpec spec; + + public FeatureRowDecoder(String featureSetRef, FeatureSetSpec spec) { + this.featureSetRef = featureSetRef; + this.spec = spec; + } + + /** + * A feature row is considered encoded if the feature set and field names are not set. This method + * is required for backward compatibility purposes, to allow Feast serving to continue serving non + * encoded Feature Row ingested by an older version of Feast. + * + * @param featureRow Feature row + * @return boolean + */ + public Boolean isEncoded(FeatureRow featureRow) { + return featureRow.getFeatureSet().isEmpty() + && featureRow.getFieldsList().stream().allMatch(field -> field.getName().isEmpty()); + } + + /** + * Validates if an encoded feature row can be decoded without exception. + * + * @param featureRow Feature row + * @return boolean + */ + public Boolean isEncodingValid(FeatureRow featureRow) { + return featureRow.getFieldsList().size() == spec.getFeaturesList().size(); + } + + /** + * Decoding feature row by repopulating the field names based on the corresponding feature set + * spec. + * + * @param encodedFeatureRow Feature row + * @return boolean + */ + public FeatureRow decode(FeatureRow encodedFeatureRow) { + final List fieldsWithoutName = encodedFeatureRow.getFieldsList(); + + List featureNames = + spec.getFeaturesList().stream() + .sorted(Comparator.comparing(FeatureSpec::getName)) + .map(FeatureSpec::getName) + .collect(Collectors.toList()); + List fields = + IntStream.range(0, featureNames.size()) + .mapToObj( + featureNameIndex -> { + String featureName = featureNames.get(featureNameIndex); + return fieldsWithoutName + .get(featureNameIndex) + .toBuilder() + .setName(featureName) + .build(); + }) + .collect(Collectors.toList()); + return encodedFeatureRow + .toBuilder() + .clearFields() + .setFeatureSet(featureSetRef) + .addAllFields(fields) + .build(); + } +} diff --git a/serving/src/main/java/feast/serving/service/RedisBackedJobService.java b/serving/src/main/java/feast/serving/service/RedisBackedJobService.java index 230e20cd782..0bf53630379 100644 --- a/serving/src/main/java/feast/serving/service/RedisBackedJobService.java +++ b/serving/src/main/java/feast/serving/service/RedisBackedJobService.java @@ -19,12 +19,11 @@ import com.google.protobuf.util.JsonFormat; import feast.serving.ServingAPIProto.Job; import feast.serving.ServingAPIProto.Job.Builder; +import io.lettuce.core.api.StatefulRedisConnection; +import io.lettuce.core.api.sync.RedisCommands; import java.util.Optional; import org.joda.time.Duration; import org.slf4j.Logger; -import redis.clients.jedis.Jedis; -import redis.clients.jedis.JedisPool; -import redis.clients.jedis.exceptions.JedisConnectionException; // TODO: Do rate limiting, currently if clients call get() or upsert() // and an exceedingly high rate e.g. they wrap job reload in a while loop with almost no wait @@ -33,53 +32,41 @@ public class RedisBackedJobService implements JobService { private static final Logger log = org.slf4j.LoggerFactory.getLogger(RedisBackedJobService.class); - private final JedisPool jedisPool; + private final RedisCommands syncCommand; // Remove job state info after "defaultExpirySeconds" to prevent filling up Redis memory // and since users normally don't require info about relatively old jobs. private final int defaultExpirySeconds = (int) Duration.standardDays(1).getStandardSeconds(); - public RedisBackedJobService(JedisPool jedisPool) { - this.jedisPool = jedisPool; + public RedisBackedJobService(StatefulRedisConnection connection) { + this.syncCommand = connection.sync(); } @Override public Optional get(String id) { - Jedis jedis = null; Job job = null; try { - jedis = jedisPool.getResource(); - String json = jedis.get(id); - if (json == null) { + String json = new String(syncCommand.get(id.getBytes())); + if (json.isEmpty()) { return Optional.empty(); } Builder builder = Job.newBuilder(); JsonFormat.parser().merge(json, builder); job = builder.build(); - } catch (JedisConnectionException e) { - log.error(String.format("Failed to connect to the redis instance: %s", e)); } catch (Exception e) { log.error(String.format("Failed to parse JSON for Feast job: %s", e.getMessage())); - } finally { - if (jedis != null) { - jedis.close(); - } } return Optional.ofNullable(job); } @Override public void upsert(Job job) { - Jedis jedis = null; try { - jedis = jedisPool.getResource(); - jedis.set(job.getId(), JsonFormat.printer().omittingInsignificantWhitespace().print(job)); - jedis.expire(job.getId(), defaultExpirySeconds); + syncCommand.set( + job.getId().getBytes(), + JsonFormat.printer().omittingInsignificantWhitespace().print(job).getBytes()); + syncCommand.expire(job.getId().getBytes(), defaultExpirySeconds); } catch (Exception e) { log.error(String.format("Failed to upsert job: %s", e.getMessage())); - } finally { - if (jedis != null) { - jedis.close(); - } } } } diff --git a/serving/src/main/java/feast/serving/service/RedisServingService.java b/serving/src/main/java/feast/serving/service/RedisServingService.java index 24c69b9f796..b298747c322 100644 --- a/serving/src/main/java/feast/serving/service/RedisServingService.java +++ b/serving/src/main/java/feast/serving/service/RedisServingService.java @@ -16,6 +16,7 @@ */ package feast.serving.service; +import static feast.serving.util.Metrics.invalidEncodingCount; import static feast.serving.util.Metrics.missingKeyCount; import static feast.serving.util.Metrics.requestCount; import static feast.serving.util.Metrics.requestLatency; @@ -41,6 +42,7 @@ import feast.serving.ServingAPIProto.GetOnlineFeaturesRequest.EntityRow; import feast.serving.ServingAPIProto.GetOnlineFeaturesResponse; import feast.serving.ServingAPIProto.GetOnlineFeaturesResponse.FieldValues; +import feast.serving.encoding.FeatureRowDecoder; import feast.serving.specs.CachedSpecService; import feast.serving.specs.FeatureSetRequest; import feast.serving.util.RefUtil; @@ -49,24 +51,28 @@ import feast.types.FieldProto.Field; import feast.types.ValueProto.Value; import io.grpc.Status; +import io.lettuce.core.api.StatefulRedisConnection; +import io.lettuce.core.api.sync.RedisCommands; import io.opentracing.Scope; import io.opentracing.Tracer; import java.util.List; import java.util.Map; +import java.util.concurrent.ExecutionException; import java.util.stream.Collectors; import org.slf4j.Logger; -import redis.clients.jedis.Jedis; -import redis.clients.jedis.JedisPool; public class RedisServingService implements ServingService { private static final Logger log = org.slf4j.LoggerFactory.getLogger(RedisServingService.class); - private final JedisPool jedisPool; private final CachedSpecService specService; private final Tracer tracer; + private final RedisCommands syncCommands; - public RedisServingService(JedisPool jedisPool, CachedSpecService specService, Tracer tracer) { - this.jedisPool = jedisPool; + public RedisServingService( + StatefulRedisConnection connection, + CachedSpecService specService, + Tracer tracer) { + this.syncCommands = connection.sync(); this.specService = specService; this.tracer = tracer; } @@ -106,7 +112,7 @@ public GetOnlineFeaturesResponse getOnlineFeatures(GetOnlineFeaturesRequest requ try { sendAndProcessMultiGet(redisKeys, entityRows, featureValuesMap, featureSetRequest); - } catch (InvalidProtocolBufferException e) { + } catch (InvalidProtocolBufferException | ExecutionException e) { throw Status.INTERNAL .withDescription("Unable to parse protobuf while retrieving feature") .withCause(e) @@ -192,9 +198,9 @@ private void sendAndProcessMultiGet( List entityRows, Map> featureValuesMap, FeatureSetRequest featureSetRequest) - throws InvalidProtocolBufferException { + throws InvalidProtocolBufferException, ExecutionException { - List jedisResps = sendMultiGet(redisKeys); + List values = sendMultiGet(redisKeys); long startTime = System.currentTimeMillis(); try (Scope scope = tracer.buildSpan("Redis-processResponse").startActive(true)) { FeatureSetSpec spec = featureSetRequest.getSpec(); @@ -206,12 +212,12 @@ private void sendAndProcessMultiGet( RefUtil::generateFeatureStringRef, featureReference -> Value.newBuilder().build())); - for (int i = 0; i < jedisResps.size(); i++) { + for (int i = 0; i < values.size(); i++) { EntityRow entityRow = entityRows.get(i); Map featureValues = featureValuesMap.get(entityRow); - byte[] jedisResponse = jedisResps.get(i); - if (jedisResponse == null) { + byte[] value = values.get(i); + if (value == null) { featureSetRequest .getFeatureReferences() .parallelStream() @@ -226,7 +232,28 @@ private void sendAndProcessMultiGet( continue; } - FeatureRow featureRow = FeatureRow.parseFrom(jedisResponse); + FeatureRow featureRow = FeatureRow.parseFrom(value); + String featureSetRef = redisKeys.get(i).getFeatureSet(); + FeatureRowDecoder decoder = + new FeatureRowDecoder(featureSetRef, specService.getFeatureSetSpec(featureSetRef)); + if (decoder.isEncoded(featureRow)) { + if (decoder.isEncodingValid(featureRow)) { + featureRow = decoder.decode(featureRow); + } else { + featureSetRequest + .getFeatureReferences() + .parallelStream() + .forEach( + request -> + invalidEncodingCount + .labels( + spec.getProject(), + String.format("%s:%d", request.getName(), request.getVersion())) + .inc()); + featureValues.putAll(nullValues); + continue; + } + } boolean stale = isStale(featureSetRequest, entityRow, featureRow); if (stale) { @@ -298,13 +325,15 @@ private boolean isStale( private List sendMultiGet(List keys) { try (Scope scope = tracer.buildSpan("Redis-sendMultiGet").startActive(true)) { long startTime = System.currentTimeMillis(); - try (Jedis jedis = jedisPool.getResource()) { + try { byte[][] binaryKeys = keys.stream() .map(AbstractMessageLite::toByteArray) .collect(Collectors.toList()) .toArray(new byte[0][0]); - return jedis.mget(binaryKeys); + return syncCommands.mget(binaryKeys).stream() + .map(keyValue -> keyValue.getValueOrElse(null)) + .collect(Collectors.toList()); } catch (Exception e) { throw Status.NOT_FOUND .withDescription("Unable to retrieve feature from Redis") diff --git a/serving/src/main/java/feast/serving/specs/CachedSpecService.java b/serving/src/main/java/feast/serving/specs/CachedSpecService.java index 35119589b27..12a8242da13 100644 --- a/serving/src/main/java/feast/serving/specs/CachedSpecService.java +++ b/serving/src/main/java/feast/serving/specs/CachedSpecService.java @@ -90,6 +90,7 @@ public CachedSpecService(CoreSpecService coreService, Path configPath) { featureSetCacheLoader = CacheLoader.from(featureSets::get); featureSetCache = CacheBuilder.newBuilder().maximumSize(MAX_SPEC_COUNT).build(featureSetCacheLoader); + featureSetCache.putAll(featureSets); } /** @@ -101,6 +102,10 @@ public Store getStore() { return this.store; } + public FeatureSetSpec getFeatureSetSpec(String featureSetRef) throws ExecutionException { + return featureSetCache.get(featureSetRef); + } + /** * Get FeatureSetSpecs for the given features. * diff --git a/serving/src/main/java/feast/serving/util/Metrics.java b/serving/src/main/java/feast/serving/util/Metrics.java index 05546ec384b..49202e2e7af 100644 --- a/serving/src/main/java/feast/serving/util/Metrics.java +++ b/serving/src/main/java/feast/serving/util/Metrics.java @@ -46,6 +46,14 @@ public class Metrics { .labelNames("project", "feature_name") .register(); + public static final Counter invalidEncodingCount = + Counter.build() + .name("invalid_encoding_feature_count") + .subsystem("feast_serving") + .help("number requested feature rows that were stored with the wrong encoding") + .labelNames("project", "feature_name") + .register(); + public static final Counter staleKeyCount = Counter.build() .name("stale_feature_count") diff --git a/serving/src/test/java/feast/serving/encoding/FeatureRowDecoderTest.java b/serving/src/test/java/feast/serving/encoding/FeatureRowDecoderTest.java new file mode 100644 index 00000000000..8f6c79ad66c --- /dev/null +++ b/serving/src/test/java/feast/serving/encoding/FeatureRowDecoderTest.java @@ -0,0 +1,110 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2020 The Feast Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package feast.serving.encoding; + +import static org.junit.Assert.*; + +import com.google.protobuf.Timestamp; +import feast.core.FeatureSetProto; +import feast.core.FeatureSetProto.FeatureSetSpec; +import feast.types.FeatureRowProto; +import feast.types.FieldProto.Field; +import feast.types.ValueProto.Value; +import feast.types.ValueProto.ValueType; +import java.util.Collections; +import org.junit.Test; + +public class FeatureRowDecoderTest { + + private FeatureSetProto.EntitySpec entity = + FeatureSetProto.EntitySpec.newBuilder().setName("entity1").build(); + + private FeatureSetSpec spec = + FeatureSetSpec.newBuilder() + .addAllEntities(Collections.singletonList(entity)) + .addFeatures( + FeatureSetProto.FeatureSpec.newBuilder() + .setName("feature1") + .setValueType(ValueType.Enum.FLOAT)) + .addFeatures( + FeatureSetProto.FeatureSpec.newBuilder() + .setName("feature2") + .setValueType(ValueType.Enum.INT32)) + .setName("feature_set_name") + .build(); + + @Test + public void featureRowWithFieldNamesIsNotConsideredAsEncoded() { + + FeatureRowDecoder decoder = new FeatureRowDecoder("feature_set_ref", spec); + FeatureRowProto.FeatureRow nonEncodedFeatureRow = + FeatureRowProto.FeatureRow.newBuilder() + .setFeatureSet("feature_set_ref") + .setEventTimestamp(Timestamp.newBuilder().setNanos(1000)) + .addFields( + Field.newBuilder().setName("feature1").setValue(Value.newBuilder().setInt32Val(2))) + .addFields( + Field.newBuilder() + .setName("feature2") + .setValue(Value.newBuilder().setFloatVal(1.0f))) + .build(); + assertFalse(decoder.isEncoded(nonEncodedFeatureRow)); + } + + @Test + public void encodingIsInvalidIfNumberOfFeaturesInSpecDiffersFromFeatureRow() { + + FeatureRowDecoder decoder = new FeatureRowDecoder("feature_set_ref", spec); + + FeatureRowProto.FeatureRow encodedFeatureRow = + FeatureRowProto.FeatureRow.newBuilder() + .setEventTimestamp(Timestamp.newBuilder().setNanos(1000)) + .addFields(Field.newBuilder().setValue(Value.newBuilder().setInt32Val(2))) + .build(); + + assertFalse(decoder.isEncodingValid(encodedFeatureRow)); + } + + @Test + public void shouldDecodeValidEncodedFeatureRow() { + + FeatureRowDecoder decoder = new FeatureRowDecoder("feature_set_ref", spec); + + FeatureRowProto.FeatureRow encodedFeatureRow = + FeatureRowProto.FeatureRow.newBuilder() + .setEventTimestamp(Timestamp.newBuilder().setNanos(1000)) + .addFields(Field.newBuilder().setValue(Value.newBuilder().setInt32Val(2))) + .addFields(Field.newBuilder().setValue(Value.newBuilder().setFloatVal(1.0f))) + .build(); + + FeatureRowProto.FeatureRow expectedFeatureRow = + FeatureRowProto.FeatureRow.newBuilder() + .setFeatureSet("feature_set_ref") + .setEventTimestamp(Timestamp.newBuilder().setNanos(1000)) + .addFields( + Field.newBuilder().setName("feature1").setValue(Value.newBuilder().setInt32Val(2))) + .addFields( + Field.newBuilder() + .setName("feature2") + .setValue(Value.newBuilder().setFloatVal(1.0f))) + .build(); + + assertTrue(decoder.isEncoded(encodedFeatureRow)); + assertTrue(decoder.isEncodingValid(encodedFeatureRow)); + assertEquals(expectedFeatureRow, decoder.decode(encodedFeatureRow)); + } +} diff --git a/serving/src/test/java/feast/serving/service/RedisBackedJobServiceTest.java b/serving/src/test/java/feast/serving/service/RedisBackedJobServiceTest.java index 9247375f59e..34bc31d2c26 100644 --- a/serving/src/test/java/feast/serving/service/RedisBackedJobServiceTest.java +++ b/serving/src/test/java/feast/serving/service/RedisBackedJobServiceTest.java @@ -16,17 +16,17 @@ */ package feast.serving.service; +import io.lettuce.core.RedisClient; +import io.lettuce.core.RedisURI; +import io.lettuce.core.codec.ByteArrayCodec; import java.io.IOException; import org.junit.After; import org.junit.Before; import org.junit.Test; -import redis.clients.jedis.JedisPool; -import redis.clients.jedis.JedisPoolConfig; import redis.embedded.RedisServer; public class RedisBackedJobServiceTest { - private static String REDIS_HOST = "localhost"; - private static int REDIS_PORT = 51235; + private static Integer REDIS_PORT = 51235; private RedisServer redis; @Before @@ -41,12 +41,10 @@ public void teardown() { } @Test - public void shouldRecoverIfRedisConnectionIsLost() { - JedisPoolConfig jedisPoolConfig = new JedisPoolConfig(); - jedisPoolConfig.setMaxTotal(1); - jedisPoolConfig.setMaxWaitMillis(10); - JedisPool jedisPool = new JedisPool(jedisPoolConfig, REDIS_HOST, REDIS_PORT); - RedisBackedJobService jobService = new RedisBackedJobService(jedisPool); + public void shouldRecoverIfRedisConnectionIsLost() throws IOException { + RedisClient client = RedisClient.create(RedisURI.create("localhost", REDIS_PORT)); + RedisBackedJobService jobService = + new RedisBackedJobService(client.connect(new ByteArrayCodec())); jobService.get("does not exist"); redis.stop(); try { @@ -56,5 +54,6 @@ public void shouldRecoverIfRedisConnectionIsLost() { } redis.start(); jobService.get("does not exist"); + client.shutdown(); } } diff --git a/serving/src/test/java/feast/serving/service/RedisServingServiceTest.java b/serving/src/test/java/feast/serving/service/RedisServingServiceTest.java index 042107e1177..05a24d3fe6a 100644 --- a/serving/src/test/java/feast/serving/service/RedisServingServiceTest.java +++ b/serving/src/test/java/feast/serving/service/RedisServingServiceTest.java @@ -38,38 +38,37 @@ import feast.types.FeatureRowProto.FeatureRow; import feast.types.FieldProto.Field; import feast.types.ValueProto.Value; +import io.lettuce.core.KeyValue; +import io.lettuce.core.api.StatefulRedisConnection; +import io.lettuce.core.api.sync.RedisCommands; import io.opentracing.Tracer; import io.opentracing.Tracer.SpanBuilder; -import java.util.Collections; -import java.util.List; -import java.util.Map; +import java.util.*; import java.util.stream.Collectors; import org.junit.Before; import org.junit.Test; import org.mockito.ArgumentMatchers; import org.mockito.Mock; import org.mockito.Mockito; -import redis.clients.jedis.Jedis; -import redis.clients.jedis.JedisPool; public class RedisServingServiceTest { - @Mock JedisPool jedisPool; - - @Mock Jedis jedis; - @Mock CachedSpecService specService; @Mock Tracer tracer; + @Mock StatefulRedisConnection connection; + + @Mock RedisCommands syncCommands; + private RedisServingService redisServingService; private byte[][] redisKeyList; @Before public void setUp() { initMocks(this); - - redisServingService = new RedisServingService(jedisPool, specService, tracer); + when(connection.sync()).thenReturn(syncCommands); + redisServingService = new RedisServingService(connection, specService, tracer); redisKeyList = Lists.newArrayList( RedisKey.newBuilder() @@ -149,12 +148,14 @@ public void shouldReturnResponseWithValuesIfKeysPresent() { .setSpec(getFeatureSetSpec()) .build(); - List featureRowBytes = - featureRows.stream().map(AbstractMessageLite::toByteArray).collect(Collectors.toList()); + List> featureRowBytes = + featureRows.stream() + .map(x -> KeyValue.from(new byte[1], Optional.of(x.toByteArray()))) + .collect(Collectors.toList()); when(specService.getFeatureSets(request.getFeaturesList())) .thenReturn(Collections.singletonList(featureSetRequest)); - when(jedisPool.getResource()).thenReturn(jedis); - when(jedis.mget(redisKeyList)).thenReturn(featureRowBytes); + when(connection.sync()).thenReturn(syncCommands); + when(syncCommands.mget(redisKeyList)).thenReturn(featureRowBytes); when(tracer.buildSpan(ArgumentMatchers.any())).thenReturn(Mockito.mock(SpanBuilder.class)); GetOnlineFeaturesResponse expected = @@ -234,12 +235,14 @@ public void shouldReturnResponseWithValuesWhenFeatureSetSpecHasUnspecifiedMaxAge .setSpec(getFeatureSetSpecWithNoMaxAge()) .build(); - List featureRowBytes = - featureRows.stream().map(AbstractMessageLite::toByteArray).collect(Collectors.toList()); + List> featureRowBytes = + featureRows.stream() + .map(x -> KeyValue.from(new byte[1], Optional.of(x.toByteArray()))) + .collect(Collectors.toList()); when(specService.getFeatureSets(request.getFeaturesList())) .thenReturn(Collections.singletonList(featureSetRequest)); - when(jedisPool.getResource()).thenReturn(jedis); - when(jedis.mget(redisKeyList)).thenReturn(featureRowBytes); + when(connection.sync()).thenReturn(syncCommands); + when(syncCommands.mget(redisKeyList)).thenReturn(featureRowBytes); when(tracer.buildSpan(ArgumentMatchers.any())).thenReturn(Mockito.mock(SpanBuilder.class)); GetOnlineFeaturesResponse expected = @@ -315,12 +318,14 @@ public void shouldReturnKeysWithoutVersionifNotProvided() { .setSpec(getFeatureSetSpec()) .build(); - List featureRowBytes = - featureRows.stream().map(AbstractMessageLite::toByteArray).collect(Collectors.toList()); + List> featureRowBytes = + featureRows.stream() + .map(x -> KeyValue.from(new byte[1], Optional.of(x.toByteArray()))) + .collect(Collectors.toList()); when(specService.getFeatureSets(request.getFeaturesList())) .thenReturn(Collections.singletonList(featureSetRequest)); - when(jedisPool.getResource()).thenReturn(jedis); - when(jedis.mget(redisKeyList)).thenReturn(featureRowBytes); + when(connection.sync()).thenReturn(syncCommands); + when(syncCommands.mget(redisKeyList)).thenReturn(featureRowBytes); when(tracer.buildSpan(ArgumentMatchers.any())).thenReturn(Mockito.mock(SpanBuilder.class)); GetOnlineFeaturesResponse expected = @@ -372,40 +377,33 @@ public void shouldReturnResponseWithUnsetValuesIfKeysNotPresent() { .putFields("entity2", strValue("b"))) .build(); - List featureRows = - Lists.newArrayList( - FeatureRow.newBuilder() - .setEventTimestamp(Timestamp.newBuilder().setSeconds(100)) - .addAllFields( - Lists.newArrayList( - Field.newBuilder().setName("entity1").setValue(intValue(1)).build(), - Field.newBuilder().setName("entity2").setValue(strValue("a")).build(), - Field.newBuilder().setName("feature1").setValue(intValue(1)).build(), - Field.newBuilder().setName("feature2").setValue(intValue(1)).build())) - .setFeatureSet("featureSet:1") - .build(), - FeatureRow.newBuilder() - .setEventTimestamp(Timestamp.newBuilder()) - .addAllFields( - Lists.newArrayList( - Field.newBuilder().setName("entity1").setValue(intValue(2)).build(), - Field.newBuilder().setName("entity2").setValue(strValue("b")).build(), - Field.newBuilder().setName("feature1").build(), - Field.newBuilder().setName("feature2").build())) - .setFeatureSet("featureSet:1") - .build()); - FeatureSetRequest featureSetRequest = FeatureSetRequest.newBuilder() .addAllFeatureReferences(request.getFeaturesList()) .setSpec(getFeatureSetSpec()) .build(); - List featureRowBytes = Lists.newArrayList(featureRows.get(0).toByteArray(), null); + FeatureRow featureRowPresent = + FeatureRow.newBuilder() + .setEventTimestamp(Timestamp.newBuilder().setSeconds(100)) + .addAllFields( + Lists.newArrayList( + Field.newBuilder().setName("entity1").setValue(intValue(1)).build(), + Field.newBuilder().setName("entity2").setValue(strValue("a")).build(), + Field.newBuilder().setName("feature1").setValue(intValue(1)).build(), + Field.newBuilder().setName("feature2").setValue(intValue(1)).build())) + .setFeatureSet("featureSet:1") + .build(); + + List> featureRowBytes = + Lists.newArrayList( + KeyValue.from(new byte[1], Optional.of(featureRowPresent.toByteArray())), + KeyValue.from(new byte[1], Optional.empty())); + when(specService.getFeatureSets(request.getFeaturesList())) .thenReturn(Collections.singletonList(featureSetRequest)); - when(jedisPool.getResource()).thenReturn(jedis); - when(jedis.mget(redisKeyList)).thenReturn(featureRowBytes); + when(connection.sync()).thenReturn(syncCommands); + when(syncCommands.mget(redisKeyList)).thenReturn(featureRowBytes); when(tracer.buildSpan(ArgumentMatchers.any())).thenReturn(Mockito.mock(SpanBuilder.class)); GetOnlineFeaturesResponse expected = @@ -489,12 +487,14 @@ public void shouldReturnResponseWithUnsetValuesIfMaxAgeIsExceeded() { .setSpec(spec) .build(); - List featureRowBytes = - featureRows.stream().map(AbstractMessageLite::toByteArray).collect(Collectors.toList()); + List> featureRowBytes = + featureRows.stream() + .map(x -> KeyValue.from(new byte[1], Optional.of(x.toByteArray()))) + .collect(Collectors.toList()); when(specService.getFeatureSets(request.getFeaturesList())) .thenReturn(Collections.singletonList(featureSetRequest)); - when(jedisPool.getResource()).thenReturn(jedis); - when(jedis.mget(redisKeyList)).thenReturn(featureRowBytes); + when(connection.sync()).thenReturn(syncCommands); + when(syncCommands.mget(redisKeyList)).thenReturn(featureRowBytes); when(tracer.buildSpan(ArgumentMatchers.any())).thenReturn(Mockito.mock(SpanBuilder.class)); GetOnlineFeaturesResponse expected = @@ -569,12 +569,14 @@ public void shouldFilterOutUndesiredRows() { .setSpec(getFeatureSetSpec()) .build(); - List featureRowBytes = - featureRows.stream().map(AbstractMessageLite::toByteArray).collect(Collectors.toList()); + List> featureRowBytes = + featureRows.stream() + .map(x -> KeyValue.from(new byte[1], Optional.of(x.toByteArray()))) + .collect(Collectors.toList()); when(specService.getFeatureSets(request.getFeaturesList())) .thenReturn(Collections.singletonList(featureSetRequest)); - when(jedisPool.getResource()).thenReturn(jedis); - when(jedis.mget(redisKeyList)).thenReturn(featureRowBytes); + when(connection.sync()).thenReturn(syncCommands); + when(syncCommands.mget(redisKeyList)).thenReturn(featureRowBytes); when(tracer.buildSpan(ArgumentMatchers.any())).thenReturn(Mockito.mock(SpanBuilder.class)); GetOnlineFeaturesResponse expected =