What this PR does / why we need it:

batch_source is optional on KafkaSource and KinesisSource, but their from_proto checked it with if data_source.batch_source. An unset proto sub-message is still truthy, so the empty message was passed to DataSource.from_proto and raised:

ValueError: Could not identify the source type being added.

So a stream source applied without a batch source could be written to the registry but not read back (get_data_source, list_data_sources, etc.):

k = KafkaSource(
    name="k",
    timestamp_field="ts",
    message_format=JsonFormat(schema_json="a int"),
    kafka_bootstrap_servers="b:9092",
    topic="t",
)
DataSource.from_proto(k.to_proto())  # ValueError

This switches both checks to HasField("batch_source"), the same check PushSource.from_proto already uses.

Which issue(s) this PR fixes:

Same failure as #3852, which was closed without a fix.

Checks

• I've made sure the tests are passing.
• My commits are signed off (git commit -s)
• My PR title follows conventional commits format

Testing Strategy

• Unit tests
• Integration tests
• Manual tests
• Testing is not required for this change

Added a parametrized round-trip test for Kafka and Kinesis sources without a batch source; it fails on master and passes with this change. ruff format --check, ruff check and mypy pass on the changed files, and the unit suite passes apart from tests that need torch or a Java runtime, which fail the same way on master in my environment.