Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

feat: Feature/kafkaoptions kwargs #4902

Open
wants to merge 8 commits into
base: master
Choose a base branch
from
Open

feat: Feature/kafkaoptions kwargs #4902

wants to merge 8 commits into from

Conversation

zerafachris
Copy link

@zerafachris zerafachris commented Jan 6, 2025 •
edited
Loading

What this PR does / why we need it:

This is an attempt to introduce additional kafka settings to the Kafka stream as raised by #4894

Which issue(s) this PR fixes:

Not applicable.

Misc

Unfortunately, this is a partial solution and am looking for support on this.

I have made most of the necessary changes but unit tests are failing due to:

===================================================================== short test summary info ======================================================================
FAILED sdk/python/tests/unit/test_stream_feature_view.py::test_stream_feature_view_serialization - typeguard.TypeCheckError: argument "data_source" (feast.core.DataFormat_pb2.StreamFormat) is not an instance of feast.core.DataSource_pb2.DataSource
FAILED sdk/python/tests/unit/test_stream_feature_view.py::test_stream_feature_view_udfs - typeguard.TypeCheckError: argument "data_source" (feast.core.DataFormat_pb2.StreamFormat) is not an instance of feast.core.DataSource_pb2.DataSource
FAILED sdk/python/tests/unit/test_stream_feature_view.py::test_stream_feature_view_initialization_with_optional_fields_omitted - typeguard.TypeCheckError: argument "data_source" (feast.core.DataFormat_pb2.StreamFormat) is not an instance of feast.core.DataSource_pb2.DataSource
FAILED sdk/python/tests/unit/local_feast_tests/test_local_feature_store.py::test_apply_stream_feature_view[feature_store_with_local_registry] - typeguard.TypeCheckError: argument "data_source" (feast.core.DataFormat_pb2.StreamFormat) is not an instance of feast.core.DataSource_pb2.DataSource
FAILED sdk/python/tests/unit/local_feast_tests/test_local_feature_store.py::test_apply_stream_feature_view_udf[feature_store_with_local_registry] - typeguard.TypeCheckError: argument "data_source" (feast.core.DataFormat_pb2.StreamFormat) is not an instance of feast.core.DataSource_pb2.DataSource
FAILED sdk/python/tests/unit/local_feast_tests/test_local_feature_store.py::test_apply_stream_source[feature_store_with_local_registry] - typeguard.TypeCheckError: argument "data_source" (feast.core.DataFormat_pb2.StreamFormat) is not an instance of feast.core.DataSource_pb2.DataSource
FAILED sdk/python/tests/unit/test_data_sources.py::test_proto_conversion - typeguard.TypeCheckError: argument "data_source" (feast.core.DataFormat_pb2.StreamFormat) is not an instance of feast.core.DataSource_pb2.DataSource
FAILED sdk/python/tests/unit/local_feast_tests/test_local_feature_store.py::test_apply_stream_source_from_repo - typeguard.TypeCheckError: argument "data_source" (feast.core.DataFormat_pb2.StreamFormat) is not an instance of feast.core.DataSource_pb2.DataSource
FAILED sdk/python/tests/unit/local_feast_tests/test_local_feature_store.py::test_apply_feature_view_with_inline_stream_source[feature_store_with_local_registry] - typeguard.TypeCheckError: argument "data_source" (feast.core.DataFormat_pb2.StreamFormat) is not an instance of feast.core.DataSource_pb2.DataSource
FAILED sdk/python/tests/unit/local_feast_tests/test_local_feature_store.py::test_apply_feature_view_with_inline_stream_source_from_repo - AssertionError: stdout: b'No project found in the repository. Using project name test5578hi0q50 defined in feature_store.yaml\nApplying changes for project tes...
=============================================== 10 failed, 519 passed, 2 skipped, 558 warnings in 234.66s (0:03:54) ================================================

Any support is greatly appreciated

@zerafachris
modify spark kafka processor to remove repeated code
29d13d0 
Signed-off-by: zerafachris PERSONAL <zerafachris@gmail.com>
@zerafachris
introduce kafka_settings to KafkaOptions
e75d483 
Signed-off-by: zerafachris PERSONAL <zerafachris@gmail.com>
@zerafachris
pre-commit changes
57e773f 
Signed-off-by: zerafachris PERSONAL <zerafachris@gmail.com>
@zerafachris
recompile python lock files and build proto
555c614 
Signed-off-by: zerafachris PERSONAL <zerafachris@gmail.com>
@zerafachris zerafachris requested a review from a team as a code owner January 6, 2025 21:11
@zerafachris
include 'StreamFormat kafka_settings = 5;' to proto definition
1ef5d0a 
Signed-off-by: zerafachris PERSONAL <zerafachris@gmail.com>
@zerafachris
Copy link
Author

With commit "1ef5d0afb35e3dd2f9f8aaee6da0a28f88982a82" I was able to add the proto definition for kafka_settings as a StreamFormat. Not sure if this is correct of not for dict[str, str]

@zerafachris
resolve conflict
897df59 
Signed-off-by: zerafachris PERSONAL <zerafachris@gmail.com>
@zerafachris zerafachris mentioned this pull request Jan 6, 2025
Open
@zerafachris zerafachris changed the title WIP: feat: Feature/kafkaoptions kwargs feat: Feature/kafkaoptions kwargs Jan 7, 2025
@franciscojavierarceo
Copy link
Member

I'll try to pull this down on Friday, sorry traveling for work.

@franciscojavierarceo
Copy link
Member

Hey @zerafachris mind taking a look? did you still need this? I was planning on cutting a release for it and wanted to make sure we get this in if needed.

@zerafachris
resolve conflicts 2
394fba0 
Signed-off-by: zerafachris PERSONAL <zerafachris@gmail.com>
@zerafachris
Copy link
Author

zerafachris commented Feb 4, 2025 •
edited
Loading

Hi @franciscojavierarceo, I resolved the conflicts. Ideally, yes, I would be keen to use this.

Unfortunately, I am still unable to fix the unit tests. The errors are:

===================================================================== short test summary info ======================================================================
FAILED sdk/python/tests/unit/test_stream_feature_view.py::test_stream_feature_view_serialization - typeguard.TypeCheckError: argument "data_source" (feast.core.DataFormat_pb2.StreamFormat) is not an instance of feast.core.DataSource_pb2.DataSource
FAILED sdk/python/tests/unit/test_stream_feature_view.py::test_stream_feature_view_udfs - typeguard.TypeCheckError: argument "data_source" (feast.core.DataFormat_pb2.StreamFormat) is not an instance of feast.core.DataSource_pb2.DataSource
FAILED sdk/python/tests/unit/test_stream_feature_view.py::test_stream_feature_view_initialization_with_optional_fields_omitted - typeguard.TypeCheckError: argument "data_source" (feast.core.DataFormat_pb2.StreamFormat) is not an instance of feast.core.DataSource_pb2.DataSource
FAILED sdk/python/tests/unit/local_feast_tests/test_local_feature_store.py::test_apply_stream_feature_view[feature_store_with_local_registry] - typeguard.TypeCheckError: argument "data_source" (feast.core.DataFormat_pb2.StreamFormat) is not an instance of feast.core.DataSource_pb2.DataSource
FAILED sdk/python/tests/unit/local_feast_tests/test_local_feature_store.py::test_apply_stream_feature_view_udf[feature_store_with_local_registry] - typeguard.TypeCheckError: argument "data_source" (feast.core.DataFormat_pb2.StreamFormat) is not an instance of feast.core.DataSource_pb2.DataSource
FAILED sdk/python/tests/unit/local_feast_tests/test_local_feature_store.py::test_apply_stream_source[feature_store_with_local_registry] - typeguard.TypeCheckError: argument "data_source" (feast.core.DataFormat_pb2.StreamFormat) is not an instance of feast.core.DataSource_pb2.DataSource
FAILED sdk/python/tests/unit/local_feast_tests/test_local_feature_store.py::test_apply_stream_source_from_repo - typeguard.TypeCheckError: argument "data_source" (feast.core.DataFormat_pb2.StreamFormat) is not an instance of feast.core.DataSource_pb2.DataSource
FAILED sdk/python/tests/unit/test_data_sources.py::test_proto_conversion - typeguard.TypeCheckError: argument "data_source" (feast.core.DataFormat_pb2.StreamFormat) is not an instance of feast.core.DataSource_pb2.DataSource
FAILED sdk/python/tests/unit/online_store/test_online_retrieval.py::test_get_online_features_milvus - feast.errors.FeastModuleImportError: Could not import module 'feast.infra.online_stores.milvus_online_store.milvus' while attempting to load class 'MilvusOnlin...
FAILED sdk/python/tests/unit/online_store/test_online_retrieval.py::test_milvus_lite_get_online_documents_v2 - feast.errors.FeastModuleImportError: Could not import module 'feast.infra.online_stores.milvus_online_store.milvus' while attempting to load class 'MilvusOnlin...
FAILED sdk/python/tests/unit/online_store/test_online_retrieval.py::test_milvus_native_from_feast_data - ModuleNotFoundError: No module named 'pymilvus'
FAILED sdk/python/tests/unit/local_feast_tests/test_local_feature_store.py::test_apply_feature_view_with_inline_stream_source[feature_store_with_local_registry] - typeguard.TypeCheckError: argument "data_source" (feast.core.DataFormat_pb2.StreamFormat) is not an instance of feast.core.DataSource_pb2.DataSource
FAILED sdk/python/tests/unit/local_feast_tests/test_local_feature_store.py::test_apply_feature_view_with_inline_stream_source_from_repo - AssertionError: stdout: b'No project found in the repository. Using project name testht2m82xvdq defined in feature_store.yaml\nApplying changes for project tes...
=============================================== 13 failed, 520 passed, 3 skipped, 556 warnings in 205.98s (0:03:25) ================================================

@franciscojavierarceo
Copy link
Member

I'll take a look at this today 👍

@ntkathole
Copy link
Contributor

@zerafachris can you please sign the commit so that DCO check pass on PR and also rebase it ? I can help with the unit tests failures.

jyejare
jyejare suggested changes Mar 13, 2025
View reviewed changes
Copy link
Contributor

@jyejare jyejare left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ACK pending comments.

sdk/python/feast/protos/feast/core/DataSource_pb2.pyi
@@ -223,16 +224,20 @@ class DataSource(google.protobuf.message.Message):
@property
def watermark_delay_threshold(self) -> google.protobuf.duration_pb2.Duration:
"""Watermark delay threshold for stream data"""
@property
def kafka_settings(self) -> feast.core.DataFormat_pb2.StreamFormat:
Copy link
Contributor

@jyejare jyejare Mar 13, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The unit tests are failing with the kafka_settings being of type feast.core.DataFormat_pb2.StreamFormat. The from_proto function here expects the kafka_settings (and as a data source config ) of type feast.core.DataFormat_pb2.DataSource.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If the code is rebased, the link mentioned here may point to a different code.

So being particular here:
The here link points to a kafka_settings parameter of the from_proto function in the module sdk/python/feast/data_source.py and class KafkaSource.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Reviewers

@franciscojavierarceo franciscojavierarceo franciscojavierarceo left review comments

@jyejare jyejare jyejare requested changes

At least 1 approving review is required to merge this pull request.

Assignees
No one assigned
Labels
ok-to-test
Projects
None yet
Milestone
No milestone
Development

Successfully merging this pull request may close these issues.
5 participants
@zerafachris @franciscojavierarceo @ntkathole @jyejare @lokeshrangineni