Skip to content

Commit 6589a2e

Browse files
felixwang9817felixwang9817
committed
Deprecate bootstrap_servers parameter in KafkaSource
Signed-off-by: Felix Wang <wangfelix98@gmail.com>
1 parent 19aedc2 commit 6589a2e

6 files changed

Lines changed: 57 additions & 45 deletions

File tree

‎protos/feast/core/DataSource.proto‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -128,7 +128,7 @@ message DataSource {
128128
// Java Protobuf class at the given class path
129129
message KafkaOptions {
130130
// Comma separated list of Kafka bootstrap servers. Used for feature tables without a defined source host[:port]]
131-
string bootstrap_servers = 1;
131+
string kafka_bootstrap_servers = 1;
132132

133133
// Kafka topic to collect feature data from.
134134
string topic = 2;

‎sdk/python/feast/data_source.py‎

Lines changed: 45 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -50,12 +50,12 @@ class KafkaOptions:
5050

5151
def __init__(
5252
self,
53-
bootstrap_servers: str,
53+
kafka_bootstrap_servers: str,
5454
message_format: StreamFormat,
5555
topic: str,
5656
watermark: Optional[timedelta] = None,
5757
):
58-
self.bootstrap_servers = bootstrap_servers
58+
self.kafka_bootstrap_servers = kafka_bootstrap_servers
5959
self.message_format = message_format
6060
self.topic = topic
6161
self.watermark = watermark or None
@@ -79,7 +79,7 @@ def from_proto(cls, kafka_options_proto: DataSourceProto.KafkaOptions):
7979
else kafka_options_proto.watermark.ToTimedelta()
8080
)
8181
kafka_options = cls(
82-
bootstrap_servers=kafka_options_proto.bootstrap_servers,
82+
kafka_bootstrap_servers=kafka_options_proto.kafka_bootstrap_servers,
8383
message_format=StreamFormat.from_proto(kafka_options_proto.message_format),
8484
topic=kafka_options_proto.topic,
8585
watermark=watermark,
@@ -100,7 +100,7 @@ def to_proto(self) -> DataSourceProto.KafkaOptions:
100100
watermark_duration.FromTimedelta(self.watermark)
101101

102102
kafka_options_proto = DataSourceProto.KafkaOptions(
103-
bootstrap_servers=self.bootstrap_servers,
103+
kafka_bootstrap_servers=self.kafka_bootstrap_servers,
104104
message_format=self.message_format.to_proto(),
105105
topic=self.topic,
106106
watermark=watermark_duration,
@@ -364,20 +364,13 @@ def get_table_query_string(self) -> str:
364364

365365

366366
class KafkaSource(DataSource):
367-
def validate(self, config: RepoConfig):
368-
pass
369-
370-
def get_table_column_names_and_types(
371-
self, config: RepoConfig
372-
) -> Iterable[Tuple[str, str]]:
373-
pass
374-
375367
def __init__(
376368
self,
377369
*args,
378370
name: Optional[str] = None,
379371
event_timestamp_column: Optional[str] = "",
380372
bootstrap_servers: Optional[str] = None,
373+
kafka_bootstrap_servers: Optional[str] = None,
381374
message_format: Optional[StreamFormat] = None,
382375
topic: Optional[str] = None,
383376
created_timestamp_column: Optional[str] = "",
@@ -391,28 +384,30 @@ def __init__(
391384
watermark: Optional[timedelta] = None,
392385
):
393386
"""
394-
Creates a KafkaSource stream source object.
387+
Creates a KafkaSource object.
388+
395389
Args:
396-
name: str. Name of data source, which should be unique within a project
397-
event_timestamp_column (optional): str. (Deprecated) Event timestamp column used for point in time
390+
name: Name of data source, which should be unique within a project
391+
event_timestamp_column: (Deprecated) Event timestamp column used for point in time
398392
joins of feature values.
399-
bootstrap_servers: str. The servers of the kafka broker in the form "localhost:9092".
400-
message_format: StreamFormat. StreamFormat of serialized messages.
401-
topic: str. The name of the topic to read from in the kafka source.
402-
created_timestamp_column (optional): str. Timestamp column indicating when the row
393+
bootstrap_servers: (Deprecated) The servers of the kafka broker in the form "localhost:9092".
394+
kafka_bootstrap_servers: The servers of the kafka broker in the form "localhost:9092".
395+
message_format: StreamFormat of serialized messages.
396+
topic: The name of the topic to read from in the kafka source.
397+
created_timestamp_column (optional): Timestamp column indicating when the row
403398
was created, used for deduplicating rows.
404-
field_mapping (optional): dict(str, str). A dictionary mapping of column names in this data
399+
field_mapping (optional): A dictionary mapping of column names in this data
405400
source to feature names in a feature table or view. Only used for feature
406401
columns, not entity or timestamp columns.
407-
date_partition_column (optional): str. Timestamp column used for partitioning.
408-
description (optional): str. A human-readable description.
409-
tags (optional): dict(str, str). A dictionary of key-value pairs to store arbitrary metadata.
410-
owner (optional): str. The owner of the data source, typically the email of the primary
402+
date_partition_column (optional): Timestamp column used for partitioning.
403+
description (optional): A human-readable description.
404+
tags (optional): A dictionary of key-value pairs to store arbitrary metadata.
405+
owner (optional): The owner of the data source, typically the email of the primary
411406
maintainer.
412-
timestamp_field (optional): str. Event timestamp field used for point
407+
timestamp_field (optional): Event timestamp field used for point
413408
in time joins of feature values.
414-
batch_source: DataSource. The datasource that acts as a batch source.
415-
watermark: timedelta. The watermark for stream data. Specifically how late stream data can arrive without being discarded.
409+
batch_source: The datasource that acts as a batch source.
410+
watermark: The watermark for stream data. Specifically how late stream data can arrive without being discarded.
416411
"""
417412
positional_attributes = [
418413
"name",
@@ -423,10 +418,19 @@ def __init__(
423418
]
424419
_name = name
425420
_event_timestamp_column = event_timestamp_column
426-
_bootstrap_servers = bootstrap_servers or ""
421+
_kafka_bootstrap_servers = kafka_bootstrap_servers or bootstrap_servers or ""
427422
_message_format = message_format
428423
_topic = topic or ""
429424

425+
if bootstrap_servers:
426+
warnings.warn(
427+
(
428+
"The 'bootstrap_servers' parameter has been deprecated in favor of 'kafka_bootstrap_servers'. "
429+
"Feast 0.25 and onwards will not support the 'bootstrap_servers' parameter."
430+
),
431+
DeprecationWarning,
432+
)
433+
430434
if args:
431435
warnings.warn(
432436
(
@@ -445,7 +449,7 @@ def __init__(
445449
if len(args) >= 2:
446450
_event_timestamp_column = args[1]
447451
if len(args) >= 3:
448-
_bootstrap_servers = args[2]
452+
_kafka_bootstrap_servers = args[2]
449453
if len(args) >= 4:
450454
_message_format = args[3]
451455
if len(args) >= 5:
@@ -471,7 +475,7 @@ def __init__(
471475
self.batch_source = batch_source
472476

473477
self.kafka_options = KafkaOptions(
474-
bootstrap_servers=_bootstrap_servers,
478+
kafka_bootstrap_servers=_kafka_bootstrap_servers,
475479
message_format=_message_format,
476480
topic=_topic,
477481
watermark=watermark,
@@ -487,8 +491,8 @@ def __eq__(self, other):
487491
return False
488492

489493
if (
490-
self.kafka_options.bootstrap_servers
491-
!= other.kafka_options.bootstrap_servers
494+
self.kafka_options.kafka_bootstrap_servers
495+
!= other.kafka_options.kafka_bootstrap_servers
492496
or self.kafka_options.message_format != other.kafka_options.message_format
493497
or self.kafka_options.topic != other.kafka_options.topic
494498
or self.kafka_options.watermark != other.kafka_options.watermark
@@ -513,7 +517,7 @@ def from_proto(data_source: DataSourceProto):
513517
name=data_source.name,
514518
event_timestamp_column=data_source.timestamp_field,
515519
field_mapping=dict(data_source.field_mapping),
516-
bootstrap_servers=data_source.kafka_options.bootstrap_servers,
520+
kafka_bootstrap_servers=data_source.kafka_options.kafka_bootstrap_servers,
517521
message_format=StreamFormat.from_proto(
518522
data_source.kafka_options.message_format
519523
),
@@ -548,6 +552,14 @@ def to_proto(self) -> DataSourceProto:
548552
data_source_proto.batch_source.MergeFrom(self.batch_source.to_proto())
549553
return data_source_proto
550554

555+
def validate(self, config: RepoConfig):
556+
pass
557+
558+
def get_table_column_names_and_types(
559+
self, config: RepoConfig
560+
) -> Iterable[Tuple[str, str]]:
561+
pass
562+
551563
@staticmethod
552564
def source_datatype_to_feast_value_type() -> Callable[[str], ValueType]:
553565
return type_map.redshift_to_feast_value_type

‎sdk/python/tests/integration/registration/test_registry.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -315,7 +315,7 @@ def simple_udf(x: int):
315315
stream_source = KafkaSource(
316316
name="kafka",
317317
timestamp_field="event_timestamp",
318-
bootstrap_servers="",
318+
kafka_bootstrap_servers="",
319319
message_format=AvroFormat(""),
320320
topic="topic",
321321
batch_source=FileSource(path="some path"),

‎sdk/python/tests/integration/registration/test_stream_feature_view_apply.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@ def test_apply_stream_feature_view(simple_dataset_1) -> None:
2929
stream_source = KafkaSource(
3030
name="kafka",
3131
timestamp_field="event_timestamp",
32-
bootstrap_servers="",
32+
kafka_bootstrap_servers="",
3333
message_format=AvroFormat(""),
3434
topic="topic",
3535
batch_source=file_source,
@@ -93,7 +93,7 @@ def test_stream_feature_view_udf(simple_dataset_1) -> None:
9393
stream_source = KafkaSource(
9494
name="kafka",
9595
timestamp_field="event_timestamp",
96-
bootstrap_servers="",
96+
kafka_bootstrap_servers="",
9797
message_format=AvroFormat(""),
9898
topic="topic",
9999
batch_source=file_source,

‎sdk/python/tests/unit/test_data_sources.py‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -93,7 +93,7 @@ def test_default_data_source_kw_arg_warning():
9393
)
9494
assert source.name == "name"
9595
assert source.timestamp_field == "column"
96-
assert source.kafka_options.bootstrap_servers == "bootstrap_servers"
96+
assert source.kafka_options.kafka_bootstrap_servers == "bootstrap_servers"
9797
assert source.kafka_options.topic == "topic"
9898
with pytest.raises(ValueError):
9999
KafkaSource("name", "column", "bootstrap_servers", topic="topic")
@@ -145,7 +145,7 @@ def test_default_data_source_kw_arg_warning():
145145
with pytest.warns(UserWarning):
146146
source = KafkaSource(
147147
timestamp_field="column",
148-
bootstrap_servers="bootstrap_servers",
148+
kafka_bootstrap_servers="bootstrap_servers",
149149
message_format=ProtoFormat("class_path"),
150150
topic="topic",
151151
)
@@ -203,7 +203,7 @@ def test_proto_conversion():
203203

204204
kafka_source = KafkaSource(
205205
name="test_source",
206-
bootstrap_servers="test_servers",
206+
kafka_bootstrap_servers="test_servers",
207207
message_format=ProtoFormat("class_path"),
208208
topic="test_topic",
209209
timestamp_field="event_timestamp",

‎sdk/python/tests/unit/test_feature_views.py‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ def test_create_batch_feature_view():
3131
stream_source = KafkaSource(
3232
name="kafka",
3333
timestamp_field="event_timestamp",
34-
bootstrap_servers="",
34+
kafka_bootstrap_servers="",
3535
message_format=AvroFormat(""),
3636
topic="topic",
3737
batch_source=FileSource(path="some path"),
@@ -49,7 +49,7 @@ def test_create_stream_feature_view():
4949
stream_source = KafkaSource(
5050
name="kafka",
5151
timestamp_field="event_timestamp",
52-
bootstrap_servers="",
52+
kafka_bootstrap_servers="",
5353
message_format=AvroFormat(""),
5454
topic="topic",
5555
batch_source=FileSource(path="some path"),
@@ -100,7 +100,7 @@ def test_stream_feature_view_serialization():
100100
stream_source = KafkaSource(
101101
name="kafka",
102102
timestamp_field="event_timestamp",
103-
bootstrap_servers="",
103+
kafka_bootstrap_servers="",
104104
message_format=AvroFormat(""),
105105
topic="topic",
106106
batch_source=FileSource(path="some path"),
@@ -137,7 +137,7 @@ def test_stream_feature_view_udfs():
137137
stream_source = KafkaSource(
138138
name="kafka",
139139
timestamp_field="event_timestamp",
140-
bootstrap_servers="",
140+
kafka_bootstrap_servers="",
141141
message_format=AvroFormat(""),
142142
topic="topic",
143143
batch_source=FileSource(path="some path"),
@@ -183,7 +183,7 @@ def test_stream_feature_view_initialization_with_optional_fields_omitted():
183183
stream_source = KafkaSource(
184184
name="kafka",
185185
timestamp_field="event_timestamp",
186-
bootstrap_servers="",
186+
kafka_bootstrap_servers="",
187187
message_format=AvroFormat(""),
188188
topic="topic",
189189
batch_source=FileSource(path="some path"),

0 commit comments

Comments
 (0)