Fix leaking dynamodb tables in integration tests

.github/workflows/pr_integration_tests.yml

@@ -128,6 +128,7 @@ jobs:
       - name: Install dependencies
         run: make install-python-ci-dependencies
       - name: Test python
+        if: ${{ always() }}  # this will guarantee that step won't be canceled and resources won't leak
         env:
           FEAST_SERVER_DOCKER_IMAGE_TAG: ${{ needs.build-docker-image.outputs.DOCKER_IMAGE_TAG }}
           FEAST_USAGE: "False"

sdk/python/feast/infra/online_stores/dynamodb.py

@@ -11,6 +11,7 @@
 # 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.
+import logging
 from datetime import datetime
 from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
 
@@ -34,6 +35,9 @@
     raise FeastExtrasDependencyImportError("aws", str(e))
 
 
+logger = logging.getLogger(__name__)
+
+
 class DynamoDBOnlineStoreConfig(FeastConfigBaseModel):
     """Online store config for DynamoDB store"""
 
@@ -179,9 +183,17 @@ def _delete_tables_idempotent(
                     f"{config.project}.{table_instance.name}"
                 )
                 table.delete()
+                logger.info(
+                    f"Dynamo table {config.project}.{table_instance.name} was deleted"
+                )
             except ClientError as ce:
                 # If the table deletion fails with ResourceNotFoundException,
                 # it means the table has already been deleted.
                 # Otherwise, re-raise the exception
                 if ce.response["Error"]["Code"] != "ResourceNotFoundException":
                     raise
+                else:
+                    logger.warning(
+                        f"Trying to delete table that doesn't exist:"
+                        f" {config.project}.{table_instance.name}"
+                    )

sdk/python/tests/integration/registration/test_universal_types.py

@@ -1,3 +1,4 @@
+import logging
 from dataclasses import dataclass
 from datetime import datetime, timedelta
 from typing import List
@@ -18,6 +19,8 @@
 from tests.integration.feature_repos.universal.entities import driver
 from tests.integration.feature_repos.universal.feature_views import driver_feature_view
 
+logger = logging.getLogger(__name__)
+
 
 def populate_test_configs(offline: bool):
     entity_type_feature_dtypes = [
@@ -106,14 +109,19 @@ def get_fixtures(request):
         field_mapping={"ts_1": "ts"},
     )
     fv = create_feature_view(
+        request.fixturename,
         config.feature_dtype,
         config.feature_is_list,
         config.has_empty_list,
         data_source,
     )
 
     def cleanup():
-        type_test_environment.data_source_creator.teardown()
+        try:
+            type_test_environment.data_source_creator.teardown()
+        except Exception:  # noqa
+            logger.exception("DataSourceCreator teardown has failed")
+
         type_test_environment.feature_store.teardown()
 
     request.addfinalizer(cleanup)
@@ -151,7 +159,11 @@ def test_feature_get_historical_features_types_match(offline_types_test_fixtures
     environment, config, data_source, fv = offline_types_test_fixtures
     fs = environment.feature_store
     fv = create_feature_view(
-        config.feature_dtype, config.feature_is_list, config.has_empty_list, data_source
+        "get_historical_features_types_match",
+        config.feature_dtype,
+        config.feature_is_list,
+        config.has_empty_list,
+        data_source,
     )
     entity = driver()
     fs.apply([fv, entity])
@@ -197,7 +209,11 @@ def test_feature_get_historical_features_types_match(offline_types_test_fixtures
 def test_feature_get_online_features_types_match(online_types_test_fixtures):
     environment, config, data_source, fv = online_types_test_fixtures
     fv = create_feature_view(
-        config.feature_dtype, config.feature_is_list, config.has_empty_list, data_source
+        "get_online_features_types_match",
+        config.feature_dtype,
+        config.feature_is_list,
+        config.has_empty_list,
+        data_source,
     )
     fs = environment.feature_store
     features = [fv.name + ":value"]
@@ -230,7 +246,9 @@ def test_feature_get_online_features_types_match(online_types_test_fixtures):
             assert isinstance(feature, expected_dtype)
 
 
-def create_feature_view(feature_dtype, feature_is_list, has_empty_list, data_source):
+def create_feature_view(
+    name, feature_dtype, feature_is_list, has_empty_list, data_source
+):
     if feature_is_list is True:
         if feature_dtype == "int32":
             value_type = ValueType.INT32_LIST
@@ -249,7 +267,8 @@ def create_feature_view(feature_dtype, feature_is_list, has_empty_list, data_sou
             value_type = ValueType.FLOAT
         elif feature_dtype == "bool":
             value_type = ValueType.BOOL
-    return driver_feature_view(data_source, value_type=value_type,)
+
+    return driver_feature_view(data_source, name=name, value_type=value_type,)
 
 
 def assert_expected_historical_feature_types(