diff --git a/infra/docker/ci/Dockerfile b/infra/docker/ci/Dockerfile
index d7b297eda65..b4f4504b5ed 100644
--- a/infra/docker/ci/Dockerfile
+++ b/infra/docker/ci/Dockerfile
@@ -77,7 +77,7 @@ RUN PROTOC_ZIP=protoc-${PROTOC_VERSION}-linux-x86_64.zip && \
 RUN curl -sL https://aka.ms/InstallAzureCLIDeb | bash
 
 # Install kubectl
-RUN apt-get install -y kubectl=1.20.1-00
+RUN apt-get install -y kubectl=1.20.2-00
 
 # Install helm
 RUN curl -fsSL -o get_helm.sh https://raw.githubusercontent.com/helm/helm/master/scripts/get-helm-3 && \
diff --git a/sdk/python/feast/contrib/validation/ge.py b/sdk/python/feast/contrib/validation/ge.py
index 0fd370aa676..de08a82431f 100644
--- a/sdk/python/feast/contrib/validation/ge.py
+++ b/sdk/python/feast/contrib/validation/ge.py
@@ -109,11 +109,9 @@ def udf(df: pd.DataFrame) -> pd.Series:
             if check.success:
                 continue
 
-            unexpected_count = (
-                check.result["unexpected_count"]
-                if "unexpected_count" in check.result
-                else df.shape[0]
-            )
+            if check.exception_info["raised_exception"]:
+                # ToDo: probably we should mark all rows as invalid
+                continue
 
             check_kwargs = check.expectation_config.kwargs
             check_kwargs.pop("result_format", None)
@@ -126,21 +124,31 @@ def udf(df: pd.DataFrame) -> pd.Series:
                 ]
             )
 
-            reporter.increment(
-                "feast_feature_validation_check_failed",
-                value=unexpected_count,
-                tags=[
-                    f"feature_table:{os.getenv('FEAST_INGESTION_FEATURE_TABLE', 'unknown')}",
-                    f"project:{os.getenv('FEAST_INGESTION_PROJECT_NAME', 'default')}",
-                    f"check:{check_name}",
-                ],
-            )
-
-            if check.exception_info["raised_exception"]:
-                # ToDo: probably we should mark all rows as invalid
-                continue
+            if "unexpected_count" in check.result:
+                reporter.increment(
+                    "feast_feature_validation_check_failed",
+                    value=check.result["unexpected_count"],
+                    tags=[
+                        f"feature_table:{os.getenv('FEAST_INGESTION_FEATURE_TABLE', 'unknown')}",
+                        f"project:{os.getenv('FEAST_INGESTION_PROJECT_NAME', 'default')}",
+                        f"check:{check_name}",
+                    ],
+                )
 
-            valid_rows.iloc[check.result["unexpected_index_list"]] = False
+                valid_rows.iloc[check.result["unexpected_index_list"]] = False
+
+            elif "observed_value" in check.result:
+                reporter.increment(
+                    "feast_feature_validation_observed_value",
+                    value=int(
+                        check.result["observed_value"] * 100
+                    ),  # storing as decimal with precision 2
+                    tags=[
+                        f"feature_table:{os.getenv('FEAST_INGESTION_FEATURE_TABLE', 'unknown')}",
+                        f"project:{os.getenv('FEAST_INGESTION_PROJECT_NAME', 'default')}",
+                        f"check:{check_name}",
+                    ],
+                )
 
         return valid_rows
 
diff --git a/sdk/python/feast/job_service.py b/sdk/python/feast/job_service.py
index e613b16d731..786e6ea2fe6 100644
--- a/sdk/python/feast/job_service.py
+++ b/sdk/python/feast/job_service.py
@@ -8,6 +8,7 @@
 from typing import Dict, List, Tuple, cast
 
 import grpc
+from google.api_core.exceptions import FailedPrecondition
 from google.protobuf.timestamp_pb2 import Timestamp
 
 import feast
@@ -334,7 +335,10 @@ def ensure_stream_ingestion_jobs(client: feast.Client, all_projects: bool):
         logging.info(
             f"Cancelling a stream ingestion job with job_hash={job_hash} job_id={job.get_id()} status={job.get_status()}"
         )
-        job.cancel()
+        try:
+            job.cancel()
+        except FailedPrecondition as exc:
+            logging.warning(f"Job canceling failed with exception {exc}")
 
     for job_hash in job_hashes_to_start:
         # Any job that we wish to start should be among expected table refs map
diff --git a/sdk/python/feast/staging/entities.py b/sdk/python/feast/staging/entities.py
index 461dc546dc1..dbb9095a58d 100644
--- a/sdk/python/feast/staging/entities.py
+++ b/sdk/python/feast/staging/entities.py
@@ -123,7 +123,7 @@ def create_bq_view_of_joined_features_and_entities(
     view.view_query = JOIN_TEMPLATE.format(
         entities=entities_ref,
         source=source_ref,
-        entity_key=",".join([f"source.{e} = entities.{e}" for e in entity_names]),
+        entity_key=" AND ".join([f"source.{e} = entities.{e}" for e in entity_names]),
     )
     view.expires = datetime.now() + timedelta(days=1)
     bq_client.create_table(view)