diff --git a/ldclient/client.py b/ldclient/client.py
index c8941e57..f434ae74 100644
--- a/ldclient/client.py
+++ b/ldclient/client.py
@@ -22,9 +22,6 @@
     get_plugin_hooks
 )
 from ldclient.impl.client_common import secure_mode_hash as _secure_mode_hash
-from ldclient.impl.datasource.feature_requester import FeatureRequesterImpl
-from ldclient.impl.datasource.polling import PollingUpdateProcessor
-from ldclient.impl.datasource.streaming import StreamingUpdateProcessor
 from ldclient.impl.datasystem import DataAvailability, DataSystem
 from ldclient.impl.datasystem.fdv2 import FDv2
 from ldclient.impl.evaluator import Evaluator, error_reason
@@ -37,7 +34,7 @@
 from ldclient.impl.flag_tracker import FlagTrackerImpl
 from ldclient.impl.model.feature_flag import FeatureFlag
 from ldclient.impl.rwlock import ReadWriteLock
-from ldclient.impl.stubs import NullEventProcessor, NullUpdateProcessor
+from ldclient.impl.stubs import NullEventProcessor
 from ldclient.impl.util import check_uwsgi, log
 from ldclient.interfaces import (
     BigSegmentStoreStatusProvider,
@@ -245,27 +242,6 @@ def _set_event_processor(self, config):
         self._event_processor = config.event_processor_class(config)
         return None
 
-    def _make_update_processor(self, config, store, ready, diagnostic_accumulator):
-        if config.update_processor_class:
-            log.info("Using user-specified update processor: " + str(config.update_processor_class))
-            return config.update_processor_class(config, store, ready)
-
-        if config.offline or config.use_ldd:
-            return NullUpdateProcessor(config, store, ready)
-
-        if config.stream:
-            return StreamingUpdateProcessor(config, store, ready, diagnostic_accumulator)
-
-        log.info("Disabling streaming API")
-        log.warning("You should only disable the streaming API if instructed to do so by LaunchDarkly support")
-
-        if config.feature_requester_class:
-            feature_requester = config.feature_requester_class(config)
-        else:
-            feature_requester = FeatureRequesterImpl(config)  # type: FeatureRequester
-
-        return PollingUpdateProcessor(config, feature_requester, store, ready)
-
     def get_sdk_key(self) -> Optional[str]:
         """Returns the configured SDK key."""
         return self._config.sdk_key
diff --git a/ldclient/impl/datasource/async_polling.py b/ldclient/impl/datasource/async_polling.py
index 5989c9ea..7a07021c 100644
--- a/ldclient/impl/datasource/async_polling.py
+++ b/ldclient/impl/datasource/async_polling.py
@@ -9,7 +9,7 @@
 
 from ldclient.async_config import AsyncConfig
 from ldclient.impl.aio.concurrency import AsyncEvent, AsyncRepeatingTask
-from ldclient.impl.datasource.datasource_common import async_sink_or_store
+from ldclient.impl.datasource.datasource_common import require_update_sink
 from ldclient.impl.retry import (
     FailureKind,
     RetryState,
@@ -22,6 +22,7 @@
     log
 )
 from ldclient.interfaces import (
+    AsyncDataSourceUpdateSink,
     AsyncFeatureRequester,
     AsyncFeatureStore,
     AsyncUpdateProcessor,
@@ -41,7 +42,7 @@ class AsyncPollingUpdateProcessor(AsyncUpdateProcessor):
 
     def __init__(self, config: AsyncConfig, requester: AsyncFeatureRequester, store: AsyncFeatureStore, ready: AsyncEvent, retry_state: Optional[RetryState] = None):
         self._config = config
-        self._data_source_update_sink = config.data_source_update_sink
+        self._data_source_update_sink: AsyncDataSourceUpdateSink = require_update_sink(config.data_source_update_sink)
         self._requester = requester
         self._store = store
         self._ready = ready
@@ -60,8 +61,7 @@ async def stop(self):
         log.info("Stopping AsyncPollingUpdateProcessor")
         self._task.stop()
 
-        if self._data_source_update_sink is not None:
-            self._data_source_update_sink.update_status(DataSourceState.OFF, None)
+        self._data_source_update_sink.update_status(DataSourceState.OFF, None)
 
         # Do not close the transport while an in-flight request still uses it.
         try:
@@ -73,10 +73,8 @@ async def _fetch_and_store(self) -> None:
         """Makes one poll request and records the outcome on the retry state."""
         try:
             all_data = await self._requester.get_all_data()
-            await async_sink_or_store(self._data_source_update_sink, self._store).init(all_data)
-
-            if self._data_source_update_sink is not None:
-                self._data_source_update_sink.update_status(DataSourceState.VALID, None)
+            await self._data_source_update_sink.init(all_data)
+            self._data_source_update_sink.update_status(DataSourceState.VALID, None)
 
             # Report the status before signaling readiness, so a caller that
             # wakes on readiness cannot still read INITIALIZING.
@@ -105,5 +103,4 @@ async def _fetch_and_store(self) -> None:
         delay = self._retry.next_delay
         level("%s - will retry in %.1fs" % (description, delay), exc_info=stacktrace)
 
-        if self._data_source_update_sink is not None:
-            self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info)
+        self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info)
diff --git a/ldclient/impl/datasource/async_streaming.py b/ldclient/impl/datasource/async_streaming.py
index fb3c5bc5..6cda82a9 100644
--- a/ldclient/impl/datasource/async_streaming.py
+++ b/ldclient/impl/datasource/async_streaming.py
@@ -18,8 +18,8 @@
 from ldclient.impl.datasource.datasource_common import (
     STREAM_ALL_PATH,
     StreamClosedError,
-    async_sink_or_store,
-    parse_path
+    parse_path,
+    require_update_sink
 )
 from ldclient.impl.retry import (
     FailureKind,
@@ -29,6 +29,7 @@
 )
 from ldclient.impl.util import http_error_description, log, monotonic_seconds
 from ldclient.interfaces import (
+    AsyncDataSourceUpdateSink,
     AsyncUpdateProcessor,
     DataSourceErrorInfo,
     DataSourceErrorKind,
@@ -49,7 +50,7 @@ def __init__(self, config, store, ready, diagnostic_accumulator, sse_factory: Op
         if config.payload_filter_key is not None:
             self._uri += '?%s' % parse.urlencode({'filter': config.payload_filter_key})
         self._config = config
-        self._data_source_update_sink = config.data_source_update_sink
+        self._data_source_update_sink: AsyncDataSourceUpdateSink = require_update_sink(config.data_source_update_sink)
         self._store = store
         self._running = False
         self._ready = ready
@@ -123,8 +124,7 @@ async def _run(self):
                         self._record_stream_init(False)
                         self._connection_attempt_started_monotonic = None
 
-                        if self._data_source_update_sink is not None:
-                            self._data_source_update_sink.update_status(DataSourceState.VALID, None)
+                        self._data_source_update_sink.update_status(DataSourceState.VALID, None)
 
                         if not self._ready.is_set():
                             log.info("AsyncStreamingUpdateProcessor initialized ok.")
@@ -172,8 +172,7 @@ async def stop(self):
         # OFF means an explicit shutdown. No stream failure produces it. It is
         # reported before teardown, so a slow close cannot hold back the status
         # that tells a waiter to give up. The sink drops anything after OFF.
-        if self._data_source_update_sink is not None:
-            self._data_source_update_sink.update_status(DataSourceState.OFF, None)
+        self._data_source_update_sink.update_status(DataSourceState.OFF, None)
 
         # Cancel the run task before tearing down the rest, preventing _run from
         # starting a fresh SSE connection against the session we are closing.
@@ -196,7 +195,7 @@ def initialized(self):
     # Returns True if we initialized the feature store
     async def _process_message(self, msg: Event) -> bool:
         """Process a single SSE event.  Returns True on a successful ``put``."""
-        target = async_sink_or_store(self._data_source_update_sink, self._store)
+        target = self._data_source_update_sink
         if msg.event == 'put':
             all_data = json.loads(msg.data)
             init_data: Mapping[VersionedDataKind, Mapping[str, dict]] = {FEATURES: all_data['data']['flags'], SEGMENTS: all_data['data']['segments']}
@@ -270,8 +269,7 @@ async def _handle_error(self, error: Exception) -> bool:
         delay = self._retry.next_delay
         level("%s - will retry in %.1fs" % (description, delay))
 
-        if self._data_source_update_sink is not None:
-            self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info)
+        self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info)
 
         if delay > 0:
             await asyncio.sleep(delay)
diff --git a/ldclient/impl/datasource/datasource_common.py b/ldclient/impl/datasource/datasource_common.py
index a5165ab1..ce65ce29 100644
--- a/ldclient/impl/datasource/datasource_common.py
+++ b/ldclient/impl/datasource/datasource_common.py
@@ -5,20 +5,16 @@
 # currently excluded from documentation - see docs/README.md
 
 from collections import namedtuple
-from typing import Mapping, Optional, Protocol, Union, runtime_checkable
+from typing import Mapping, Optional, Protocol, TypeVar, runtime_checkable
 
 from ldclient.impl.util import _LD_ENVID_HEADER
-from ldclient.interfaces import (
-    AsyncDataSourceUpdateSink,
-    AsyncFeatureStore,
-    DataSourceUpdateSink,
-    FeatureStore
-)
 from ldclient.versioned_data_kind import FEATURES, SEGMENTS
 
 STREAM_ALL_PATH = '/all'
 FDV1_POLLING_ENDPOINT = '/sdk/latest-all'
 
+T = TypeVar('T')
+
 ParsedPath = namedtuple('ParsedPath', ['kind', 'key'])
 
 
@@ -39,32 +35,18 @@ def __init__(self):
         super().__init__("the server closed the stream connection")
 
 
-def sink_or_store(sink: Optional[DataSourceUpdateSink], store: FeatureStore) -> Union[DataSourceUpdateSink, FeatureStore]:
-    """
-    The original implementation of the data sources relied on the feature store
-    directly, which we are trying to move away from. Customers who might have
-    instantiated one directly for some reason wouldn't know they have to set
-    the config's sink manually, so we have to fall back to the store if the
-    sink isn't present.
-
-    The next major release should be able to simplify this structure and
-    remove the need for fall back to the data store because the update sink
-    should always be present.
+def require_update_sink(sink: Optional[T]) -> T:
     """
-    if sink is None:
-        return store
-
-    return sink
+    Returns the data source update sink, or raises if it is missing.
 
+    The data system puts the sink on the config before it builds a data
+    source. A data source writes all its updates to the sink, so it cannot
+    operate without one.
 
-def async_sink_or_store(sink: Optional[AsyncDataSourceUpdateSink], store: AsyncFeatureStore) -> Union[AsyncDataSourceUpdateSink, AsyncFeatureStore]:
-    """
-    The async counterpart of :func:`sink_or_store`, kept separate so each side
-    names its own sink and store types. One shared generic function accepted a
-    sync sink beside an async store without complaint.
+    :raises ValueError: if the sink is None
     """
     if sink is None:
-        return store
+        raise ValueError("config.data_source_update_sink is not set. The data system must set it before it builds a data source.")
 
     return sink
 
diff --git a/ldclient/impl/datasource/polling.py b/ldclient/impl/datasource/polling.py
index 2eacc6ad..3fb9fcc1 100644
--- a/ldclient/impl/datasource/polling.py
+++ b/ldclient/impl/datasource/polling.py
@@ -11,7 +11,7 @@
 from ldclient.config import Config
 from ldclient.impl.datasource.datasource_common import (
     record_environment_id,
-    sink_or_store
+    require_update_sink
 )
 from ldclient.impl.repeating_task import RepeatingTask
 from ldclient.impl.retry import (
@@ -52,7 +52,7 @@ class PollingUpdateProcessor(UpdateProcessor):
 
     def __init__(self, config: Config, requester: FeatureRequester, store: FeatureStore, ready: Event, retry_state: Optional[RetryState] = None):
         self._config = config
-        self._data_source_update_sink: Optional[DataSourceUpdateSink] = config.data_source_update_sink
+        self._data_source_update_sink: DataSourceUpdateSink = require_update_sink(config.data_source_update_sink)
         self._requester = requester
         self._store = store
         self._ready = ready
@@ -69,10 +69,6 @@ def initialized(self):
     def stop(self):
         log.info("Stopping PollingUpdateProcessor")
         self._task.stop()
-
-        if self._data_source_update_sink is None:
-            return
-
         self._data_source_update_sink.update_status(DataSourceState.OFF, None)
 
     def _poll(self) -> None:
@@ -80,10 +76,8 @@ def _poll(self) -> None:
         try:
             (all_data, headers) = self._get_all_data_with_headers()
             record_environment_id(self._data_source_update_sink, headers)
-            sink_or_store(self._data_source_update_sink, self._store).init(all_data)
-
-            if self._data_source_update_sink is not None:
-                self._data_source_update_sink.update_status(DataSourceState.VALID, None)
+            self._data_source_update_sink.init(all_data)
+            self._data_source_update_sink.update_status(DataSourceState.VALID, None)
 
             # Report the status before signaling readiness, so a caller that
             # wakes on readiness cannot still read INITIALIZING.
@@ -112,8 +106,7 @@ def _poll(self) -> None:
         delay = self._retry.next_delay
         level("%s - will retry in %.1fs" % (description, delay), exc_info=stacktrace)
 
-        if self._data_source_update_sink is not None:
-            self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info)
+        self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info)
 
     def _get_all_data_with_headers(self) -> Tuple[Any, Optional[Mapping[str, str]]]:
         """
diff --git a/ldclient/impl/datasource/streaming.py b/ldclient/impl/datasource/streaming.py
index 9e425e3f..e9c30509 100644
--- a/ldclient/impl/datasource/streaming.py
+++ b/ldclient/impl/datasource/streaming.py
@@ -20,7 +20,7 @@
     StreamClosedError,
     parse_path,
     record_environment_id,
-    sink_or_store
+    require_update_sink
 )
 from ldclient.impl.http import HTTPFactory, _http_factory
 from ldclient.impl.retry import (
@@ -34,6 +34,7 @@
     DataSourceErrorInfo,
     DataSourceErrorKind,
     DataSourceState,
+    DataSourceUpdateSink,
     UpdateProcessor
 )
 from ldclient.versioned_data_kind import FEATURES, SEGMENTS
@@ -57,7 +58,7 @@ def __init__(self, config, store, ready, diagnostic_accumulator, retry_state: Op
         if config.payload_filter_key is not None:
             self._uri += '?%s' % parse.urlencode({'filter': config.payload_filter_key})
         self._config = config
-        self._data_source_update_sink = config.data_source_update_sink
+        self._data_source_update_sink: DataSourceUpdateSink = require_update_sink(config.data_source_update_sink)
         self._store = store
         self._running = False
         self._ready = ready
@@ -91,7 +92,7 @@ def run(self):
                     message_ok = False
                     message_handled = False
                     try:
-                        message_ok = self._process_message(sink_or_store(self._data_source_update_sink, self._store), action)
+                        message_ok = self._process_message(self._data_source_update_sink, action)
                         message_handled = True
                     except json.decoder.JSONDecodeError as e:
                         log.info("Error while handling stream event; will restart stream: %s" % e)
@@ -113,8 +114,7 @@ def run(self):
                         self._record_stream_init(False)
                         self._connection_attempt_started_monotonic = None
 
-                        if self._data_source_update_sink is not None:
-                            self._data_source_update_sink.update_status(DataSourceState.VALID, None)
+                        self._data_source_update_sink.update_status(DataSourceState.VALID, None)
 
                         if not self._ready.is_set():
                             log.info("StreamingUpdateProcessor initialized ok.")
@@ -167,8 +167,7 @@ def stop(self):
         # OFF means an explicit shutdown. No stream failure produces it. It is
         # reported before teardown, so a slow close cannot hold back the status
         # that tells a waiter to give up. The sink drops anything after OFF.
-        if self._data_source_update_sink is not None:
-            self._data_source_update_sink.update_status(DataSourceState.OFF, None)
+        self._data_source_update_sink.update_status(DataSourceState.OFF, None)
 
         if self._sse:
             self._sse.close()
@@ -256,8 +255,7 @@ def _handle_error(self, error: Exception) -> bool:
         delay = self._retry.next_delay
         level("%s - will retry in %.1fs" % (description, delay))
 
-        if self._data_source_update_sink is not None:
-            self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info)
+        self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info)
 
         interrupted = self._stop_event.wait(min(delay, TIMEOUT_MAX))
 
diff --git a/ldclient/impl/datasystem/fdv1.py b/ldclient/impl/datasystem/fdv1.py
index 477b6e0f..008c4b03 100644
--- a/ldclient/impl/datasystem/fdv1.py
+++ b/ldclient/impl/datasystem/fdv1.py
@@ -277,7 +277,6 @@ def target_availability(self) -> DataAvailability:
         return DataAvailability.REFRESHED
 
     def _make_update_processor(self, config: Config, store: FeatureStore, ready: Event):
-        # Mirrors LDClient._make_update_processor but scoped for FDv1
         if config.update_processor_class:
             return config.update_processor_class(config, store, ready)
 
diff --git a/ldclient/impl/integrations/files/file_data_source.py b/ldclient/impl/integrations/files/file_data_source.py
index 81fb38d3..7d5e0d31 100644
--- a/ldclient/impl/integrations/files/file_data_source.py
+++ b/ldclient/impl/integrations/files/file_data_source.py
@@ -4,6 +4,7 @@
 import traceback
 from typing import Optional
 
+from ldclient.impl.datasource.datasource_common import require_update_sink
 from ldclient.impl.repeating_task import RepeatingTask
 from ldclient.impl.util import log
 from ldclient.interfaces import (
@@ -41,8 +42,7 @@ def _sanitize_json_item(item):
 
 class _FileDataSource(UpdateProcessor):
     def __init__(self, store, data_source_update_sink: Optional[DataSourceUpdateSink], ready, paths, auto_update, poll_interval, force_polling):
-        self._store = store
-        self._data_source_update_sink = data_source_update_sink
+        self._data_source_update_sink: DataSourceUpdateSink = require_update_sink(data_source_update_sink)
         self._ready = ready
         self._inited = False
         self._paths = paths
@@ -53,23 +53,6 @@ def __init__(self, store, data_source_update_sink: Optional[DataSourceUpdateSink
         self._poll_interval = poll_interval
         self._force_polling = force_polling
 
-    def _sink_or_store(self):
-        """
-        The original implementation of this class relied on the feature store
-        directly, which we are trying to move away from. Customers who might have
-        instantiated this directly for some reason wouldn't know they have to set
-        the config's sink manually, so we have to fall back to the store if the
-        sink isn't present.
-
-        The next major release should be able to simplify this structure and
-        remove the need for fall back to the data store because the update sink
-        should always be present.
-        """
-        if self._data_source_update_sink is None:
-            return self._store
-
-        return self._data_source_update_sink
-
     def start(self):
         self._load_all()
 
@@ -95,19 +78,16 @@ def _load_all(self):
             except Exception as e:
                 log.error('Unable to load flag data from "%s": %s' % (path, repr(e)))
                 traceback.print_exc()
-                if self._data_source_update_sink is not None:
-                    self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, DataSourceErrorInfo(DataSourceErrorKind.INVALID_DATA, 0, time.time, str(e)))
+                self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, DataSourceErrorInfo(DataSourceErrorKind.INVALID_DATA, 0, time.time, str(e)))
                 return
         try:
-            self._sink_or_store().init(all_data)
+            self._data_source_update_sink.init(all_data)
             self._inited = True
-            if self._data_source_update_sink is not None:
-                self._data_source_update_sink.update_status(DataSourceState.VALID, None)
+            self._data_source_update_sink.update_status(DataSourceState.VALID, None)
         except Exception as e:
             log.error('Unable to store data: %s' % repr(e))
             traceback.print_exc()
-            if self._data_source_update_sink is not None:
-                self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time, str(e)))
+            self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time, str(e)))
 
     def _load_file(self, path, all_data):
         content = None
diff --git a/ldclient/testing/impl/datasource/test_async_polling.py b/ldclient/testing/impl/datasource/test_async_polling.py
index 8d497ae0..7a70c0b3 100644
--- a/ldclient/testing/impl/datasource/test_async_polling.py
+++ b/ldclient/testing/impl/datasource/test_async_polling.py
@@ -30,7 +30,10 @@
     DataSourceErrorKind,
     DataSourceState
 )
-from ldclient.testing.mock_async_components import MockAsyncFeatureStore
+from ldclient.testing.mock_async_components import (
+    MockAsyncFeatureStore,
+    with_async_update_sink
+)
 from ldclient.testing.test_util import no_retry_jitter
 from ldclient.versioned_data_kind import FEATURES, SEGMENTS
 
@@ -86,7 +89,7 @@ def make_processor(config=None, store=None, ready=None, requester=None, retry_st
         requester = MagicMock()
         requester.close = AsyncMock()
     return AsyncPollingUpdateProcessor(
-        config=config,
+        config=with_async_update_sink(config, store),
         requester=requester,
         store=store,
         ready=ready,
@@ -707,3 +710,8 @@ async def hang():
             await stop_task
 
         requester.close.assert_awaited_once()
+
+
+def test_constructor_raises_without_update_sink():
+    with pytest.raises(ValueError, match="data_source_update_sink"):
+        AsyncPollingUpdateProcessor(make_config(), MagicMock(), MockAsyncFeatureStore(), asyncio.Event())
diff --git a/ldclient/testing/impl/datasource/test_async_streaming.py b/ldclient/testing/impl/datasource/test_async_streaming.py
index 302a2a3c..0ab1c66c 100644
--- a/ldclient/testing/impl/datasource/test_async_streaming.py
+++ b/ldclient/testing/impl/datasource/test_async_streaming.py
@@ -33,7 +33,10 @@
 )
 from ldclient.interfaces import DataSourceErrorKind, DataSourceState
 from ldclient.testing.builders import FlagBuilder, SegmentBuilder
-from ldclient.testing.mock_async_components import MockAsyncFeatureStore
+from ldclient.testing.mock_async_components import (
+    MockAsyncFeatureStore,
+    with_async_update_sink
+)
 from ldclient.testing.test_util import (
     no_retry_jitter,
     record_healthy_windows,
@@ -188,7 +191,7 @@ def _make_processor(actions, config=None, store=None, ready_event=None, diag=Non
     store = store or MockAsyncFeatureStore()
     ready_event = ready_event or asyncio.Event()
     factory = _MockSSEFactory(actions)
-    proc = AsyncStreamingUpdateProcessor(config, store, ready_event, diag, factory, retry_state=retry_state)
+    proc = AsyncStreamingUpdateProcessor(with_async_update_sink(config, store), store, ready_event, diag, factory, retry_state=retry_state)
     return proc, store, ready_event, factory
 
 
@@ -782,7 +785,7 @@ async def test_default_construction_builds_configured_session():
         async_streaming, "AsyncSSEFactory"
     ) as factory_cls:
         factory_cls.return_value.create.return_value = _MockSSE([])  # empty stream -> _run blocks
-        proc = AsyncStreamingUpdateProcessor(config, store, ready, None)
+        proc = AsyncStreamingUpdateProcessor(with_async_update_sink(config, store), store, ready, None)
 
         # Deferred: nothing is created until the task runs.
         assert proc._owned_session is None
@@ -809,7 +812,7 @@ async def test_default_construction_session_closed_on_stop():
         async_streaming, "make_client_session", return_value=fake_session
     ), mock.patch.object(async_streaming, "AsyncSSEFactory") as factory_cls:
         factory_cls.return_value.create.return_value = _MockSSE([])
-        proc = AsyncStreamingUpdateProcessor(config, store, ready, None)
+        proc = AsyncStreamingUpdateProcessor(with_async_update_sink(config, store), store, ready, None)
         proc.start()
         await _wait_until(lambda: proc._owned_session is not None)
 
@@ -832,7 +835,7 @@ async def test_default_construction_session_closed_when_run_fails():
         async_streaming, "make_client_session", return_value=fake_session
     ), mock.patch.object(async_streaming, "AsyncSSEFactory") as factory_cls:
         factory_cls.return_value.create.side_effect = RuntimeError("boom")
-        proc = AsyncStreamingUpdateProcessor(config, store, ready, None)
+        proc = AsyncStreamingUpdateProcessor(with_async_update_sink(config, store), store, ready, None)
         proc.start()
 
         # _run builds the session, then create() raises. The finally must still
@@ -933,3 +936,8 @@ def record_stream_init(self, timestamp, duration, failed):
     assert failed is False
     assert duration >= 0
     assert duration < 1000
+
+
+def test_constructor_raises_without_update_sink():
+    with pytest.raises(ValueError, match="data_source_update_sink"):
+        AsyncStreamingUpdateProcessor(_make_config(), MockAsyncFeatureStore(), asyncio.Event(), None)
diff --git a/ldclient/testing/impl/datasource/test_polling_processor.py b/ldclient/testing/impl/datasource/test_polling_processor.py
index 913438a4..9377f711 100644
--- a/ldclient/testing/impl/datasource/test_polling_processor.py
+++ b/ldclient/testing/impl/datasource/test_polling_processor.py
@@ -24,6 +24,7 @@
     DataSourceStatus
 )
 from ldclient.testing.builders import *
+from ldclient.testing.mock_components import with_update_sink
 from ldclient.testing.stub_util import MockFeatureRequester, MockResponse
 from ldclient.testing.sync_util import wait_until
 from ldclient.testing.test_util import SpyListener, no_retry_jitter
@@ -65,7 +66,7 @@ def fast_retry_state(delay=0.05):
 
 def setup_processor(config, retry_state=None):
     global pp
-    pp = PollingUpdateProcessor(config, mock_requester, store, ready, retry_state=retry_state)
+    pp = PollingUpdateProcessor(with_update_sink(config, store), mock_requester, store, ready, retry_state=retry_state)
     pp.start()
 
 
@@ -175,7 +176,7 @@ def test_the_first_success_after_an_outage_polls_at_the_cadence():
     # elapsed time.
     with no_retry_jitter():
         retry = for_polling(30)
-        processor = PollingUpdateProcessor(Config("SDK_KEY"), mock_requester, store, ready, retry_state=retry)
+        processor = PollingUpdateProcessor(with_update_sink(Config("SDK_KEY"), store), mock_requester, store, ready, retry_state=retry)
 
         mock_requester.exception = UnsuccessfulResponseException(401)
         processor._poll()
@@ -205,7 +206,7 @@ def test_transport_failures_poll_again_at_the_cadence(error):
     """No transport failure reaches the extended regime, a bad certificate
     included. Only an HTTP status can do that."""
     retry = for_polling(30)
-    processor = PollingUpdateProcessor(Config("SDK_KEY"), mock_requester, store, ready, retry_state=retry)
+    processor = PollingUpdateProcessor(with_update_sink(Config("SDK_KEY"), store), mock_requester, store, ready, retry_state=retry)
 
     mock_requester.exception = error
     processor._poll()
@@ -258,7 +259,7 @@ def test_the_log_reports_the_growing_retry_delay(caplog):
 
     with no_retry_jitter():
         retry = for_polling(30)
-        processor = PollingUpdateProcessor(Config("SDK_KEY"), mock_requester, store, ready, retry_state=retry)
+        processor = PollingUpdateProcessor(with_update_sink(Config("SDK_KEY"), store), mock_requester, store, ready, retry_state=retry)
         mock_requester.exception = UnsuccessfulResponseException(401)
         processor._poll()
         processor._poll()
@@ -277,7 +278,7 @@ def test_a_normal_failure_logs_the_poll_interval_at_warning_level(caplog):
 
     with no_retry_jitter():
         retry = for_polling(30)
-        processor = PollingUpdateProcessor(Config("SDK_KEY"), mock_requester, store, ready, retry_state=retry)
+        processor = PollingUpdateProcessor(with_update_sink(Config("SDK_KEY"), store), mock_requester, store, ready, retry_state=retry)
         mock_requester.exception = UnsuccessfulResponseException(503)
         processor._poll()
 
@@ -291,7 +292,7 @@ def test_a_transport_error_reports_a_delay_and_keeps_its_stacktrace(caplog):
 
     with no_retry_jitter():
         retry = for_polling(30)
-        processor = PollingUpdateProcessor(Config("SDK_KEY"), mock_requester, store, ready, retry_state=retry)
+        processor = PollingUpdateProcessor(with_update_sink(Config("SDK_KEY"), store), mock_requester, store, ready, retry_state=retry)
         mock_requester.exception = ConnectionResetError(104, "reset by peer")
         processor._poll()
 
@@ -357,7 +358,7 @@ def test_stop_twice_and_stop_before_start_are_safe():
     including while an hour-long wait is pending."""
     mock_requester.exception = UnsuccessfulResponseException(401)
     processor = PollingUpdateProcessor(
-        Config("SDK_KEY"), mock_requester, store, ready, retry_state=fast_retry_state(ONE_HOUR)
+        with_update_sink(Config("SDK_KEY"), store), mock_requester, store, ready, retry_state=fast_retry_state(ONE_HOUR)
     )
 
     processor.stop()
@@ -391,7 +392,7 @@ def test_a_poll_finishing_after_stop_reports_nothing():
     config = Config("SDK_KEY")
     config._data_source_update_sink = DataSourceUpdateSinkImpl(store, listeners, Listeners())
     mock_requester.all_data = {FEATURES: {}, SEGMENTS: {}}
-    processor = PollingUpdateProcessor(config, mock_requester, store, ready)
+    processor = PollingUpdateProcessor(with_update_sink(config, store), mock_requester, store, ready)
 
     processor.stop()
     processor._poll()
@@ -472,3 +473,8 @@ def test_environment_id_is_none_when_requester_provides_no_headers():
     assert ready.wait(2)
 
     assert sink.environment_id is None
+
+
+def test_constructor_raises_without_update_sink():
+    with pytest.raises(ValueError, match="data_source_update_sink"):
+        PollingUpdateProcessor(Config("SDK_KEY"), mock_requester, store, ready)
diff --git a/ldclient/testing/impl/datasource/test_streaming.py b/ldclient/testing/impl/datasource/test_streaming.py
index 0fa3b803..c153e356 100644
--- a/ldclient/testing/impl/datasource/test_streaming.py
+++ b/ldclient/testing/impl/datasource/test_streaming.py
@@ -40,6 +40,7 @@
     SequentialHandler,
     start_server
 )
+from ldclient.testing.mock_components import with_update_sink
 from ldclient.testing.proxy_test_util import do_proxy_tests
 from ldclient.testing.stub_util import (
     make_delete_event,
@@ -91,7 +92,7 @@ def test_request_properties():
             config = Config(sdk_key='sdk-key', stream_uri=server.uri)
             server.for_path('/all', stream)
 
-            with StreamingUpdateProcessor(config, store, ready, None) as sp:
+            with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None) as sp:
                 sp.start()
                 req = server.await_request()
                 assert req.method == 'GET'
@@ -112,7 +113,7 @@ def test_sends_instance_id():
             config._instance_id = 'my-instance-id'
             server.for_path('/all', stream)
 
-            with StreamingUpdateProcessor(config, store, ready, None) as sp:
+            with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None) as sp:
                 sp.start()
                 req = server.await_request()
                 assert req.headers.get('X-LaunchDarkly-Instance-Id') == 'my-instance-id'
@@ -127,7 +128,7 @@ def test_sends_wrapper_header():
             config = Config(sdk_key='sdk-key', stream_uri=server.uri, wrapper_name='Flask', wrapper_version='0.1.0')
             server.for_path('/all', stream)
 
-            with StreamingUpdateProcessor(config, store, ready, None) as sp:
+            with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None) as sp:
                 sp.start()
                 req = server.await_request()
                 assert req.headers.get('X-LaunchDarkly-Wrapper') == 'Flask/0.1.0'
@@ -142,7 +143,7 @@ def test_sends_wrapper_header_without_version():
             config = Config(sdk_key='sdk-key', stream_uri=server.uri, wrapper_name='Flask')
             server.for_path('/all', stream)
 
-            with StreamingUpdateProcessor(config, store, ready, None) as sp:
+            with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None) as sp:
                 sp.start()
                 req = server.await_request()
                 assert req.headers.get('X-LaunchDarkly-Wrapper') == 'Flask'
@@ -157,7 +158,7 @@ def test_sends_tag_header():
             config = Config(sdk_key='sdk-key', stream_uri=server.uri, application={"id": "my-id", "version": "my-version"})
             server.for_path('/all', stream)
 
-            with StreamingUpdateProcessor(config, store, ready, None) as sp:
+            with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None) as sp:
                 sp.start()
                 req = server.await_request()
                 assert req.headers.get('X-LaunchDarkly-Tags') == 'application-id/my-id application-version/my-version'
@@ -174,7 +175,7 @@ def test_receives_put_event():
             config = Config(sdk_key='sdk-key', stream_uri=server.uri)
             server.for_path('/all', stream)
 
-            with StreamingUpdateProcessor(config, store, ready, None) as sp:
+            with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None) as sp:
                 sp.start()
                 ready.wait(start_wait)
                 assert sp.initialized()
@@ -195,7 +196,7 @@ def test_receives_patch_events():
             config = Config(sdk_key='sdk-key', stream_uri=server.uri)
             server.for_path('/all', stream)
 
-            with StreamingUpdateProcessor(config, store, ready, None) as sp:
+            with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None) as sp:
                 sp.start()
                 ready.wait(start_wait)
                 assert sp.initialized()
@@ -220,7 +221,7 @@ def test_receives_delete_events():
             config = Config(sdk_key='sdk-key', stream_uri=server.uri)
             server.for_path('/all', stream)
 
-            with StreamingUpdateProcessor(config, store, ready, None) as sp:
+            with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None) as sp:
                 sp.start()
                 ready.wait(start_wait)
                 assert sp.initialized()
@@ -246,7 +247,7 @@ def test_reconnects_if_stream_is_broken():
                 config = Config(sdk_key='sdk-key', stream_uri=server.uri, initial_reconnect_delay=brief_delay)
                 server.for_path('/all', SequentialHandler(stream1, stream2))
 
-                with StreamingUpdateProcessor(config, store, ready, None) as sp:
+                with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None) as sp:
                     sp.start()
                     server.await_request
                     ready.wait(start_wait)
@@ -268,7 +269,7 @@ def test_retries_on_network_error():
             config = Config(sdk_key='sdk-key', stream_uri=server.uri, initial_reconnect_delay=brief_delay)
             server.for_path('/all', two_errors_then_success)
 
-            with StreamingUpdateProcessor(config, store, ready, None) as sp:
+            with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None) as sp:
                 sp.start()
                 ready.wait(start_wait)
                 assert sp.initialized()
@@ -287,7 +288,7 @@ def test_recoverable_http_error(status):
             config = Config(sdk_key='sdk-key', stream_uri=server.uri, initial_reconnect_delay=brief_delay)
             server.for_path('/all', two_errors_then_success)
 
-            with StreamingUpdateProcessor(config, store, ready, None) as sp:
+            with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None) as sp:
                 sp.start()
                 ready.wait(start_wait)
                 assert sp.initialized()
@@ -307,7 +308,7 @@ def test_unexpected_http_error_backs_off_a_long_way(status):
             config = Config(sdk_key='sdk-key', stream_uri=server.uri, initial_reconnect_delay=brief_delay)
             server.for_path('/all', error_then_success)
 
-            with StreamingUpdateProcessor(config, store, ready, None) as sp:
+            with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None) as sp:
                 sp.start()
                 server.wait_until_request_received()
                 # The next attempt is past the normal ceiling, so the failure
@@ -339,7 +340,7 @@ def test_unexpected_http_error_keeps_retrying(status):
             config._data_source_update_sink = DataSourceUpdateSinkImpl(store, listeners, Listeners())
             server.for_path('/all', error_then_success)
 
-            with StreamingUpdateProcessor(config, store, ready, None, retry_state=fast_retry_state()) as sp:
+            with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None, retry_state=fast_retry_state()) as sp:
                 sp.start()
                 ready.wait(start_wait)
                 assert sp.initialized()
@@ -389,7 +390,7 @@ def test_the_sdk_configures_the_sse_client_never_to_wait():
     with start_server() as server:
         server.for_path('/all', BasicResponse(503))
         config = Config(sdk_key='sdk-key', stream_uri=server.uri, initial_reconnect_delay=30)
-        sp = StreamingUpdateProcessor(config, store, Event(), None)
+        sp = StreamingUpdateProcessor(with_update_sink(config, store), store, Event(), None)
         client = sp._create_sse_client()
         try:
             actions = iter(client.all)
@@ -424,7 +425,7 @@ def test_server_close_backs_off_and_keeps_the_stream_running():
                 server.for_path('/all', SequentialHandler(stream1, stream2))
 
                 retry = fast_retry_state()
-                with StreamingUpdateProcessor(config, store, ready, None, retry_state=retry) as sp:
+                with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None, retry_state=retry) as sp:
                     sp.start()
                     ready.wait(start_wait)
                     assert sp.initialized()
@@ -447,7 +448,7 @@ def test_server_close_uses_the_normal_delay_curve():
     store = InMemoryFeatureStore()
     config = Config(sdk_key='sdk-key', initial_reconnect_delay=1)
     retry = for_streaming(1)
-    sp = StreamingUpdateProcessor(config, store, Event(), None, retry_state=retry)
+    sp = StreamingUpdateProcessor(with_update_sink(config, store), store, Event(), None, retry_state=retry)
     sp._running = True
     sp._stop_event.set()  # so the wait returns at once
 
@@ -488,7 +489,7 @@ def listener(s):
             server.for_path('/all', SequentialHandler(invalid_stream, valid_stream))
 
             retry = fast_retry_state()
-            with StreamingUpdateProcessor(config, store, ready, None, retry_state=retry) as sp:
+            with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None, retry_state=retry) as sp:
                 sp.start()
                 ready.wait(start_wait)
                 assert sp.initialized()
@@ -515,7 +516,7 @@ def test_a_leaked_interrupt_flag_does_not_swallow_a_server_close():
                 server.for_path('/all', SequentialHandler(stream1, stream2))
 
                 retry = fast_retry_state()
-                with StreamingUpdateProcessor(config, store, ready, None, retry_state=retry) as sp:
+                with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None, retry_state=retry) as sp:
                     sp._interrupted_by_sdk = True
                     sp.start()
                     ready.wait(start_wait)
@@ -530,7 +531,7 @@ def test_a_leaked_interrupt_flag_does_not_swallow_a_server_close():
 def test_stop_before_start_does_not_raise():
     """stop() can land before run() has built the SSE client."""
     config = Config(sdk_key='sdk-key', stream_uri='http://localhost')
-    sp = StreamingUpdateProcessor(config, InMemoryFeatureStore(), Event(), None)
+    sp = StreamingUpdateProcessor(with_update_sink(config, InMemoryFeatureStore()), InMemoryFeatureStore(), Event(), None)
     sp.stop()
 
 
@@ -544,7 +545,7 @@ def test_a_stop_before_the_connection_exists_still_ends_the_run():
         with stream_content(make_put_event()) as stream:
             server.for_path('/all', stream)
             config = Config(sdk_key='sdk-key', stream_uri=server.uri)
-            sp = StreamingUpdateProcessor(config, store, ready, None)
+            sp = StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None)
 
             sp.stop()
             thread = Thread(target=sp.run, daemon=True)
@@ -565,7 +566,7 @@ def test_a_raise_inside_the_loop_closes_the_connection():
         with stream_content(make_put_event()) as stream:
             server.for_path('/all', stream)
             config = Config(sdk_key='sdk-key', stream_uri=server.uri)
-            sp = StreamingUpdateProcessor(config, store, Event(), None)
+            sp = StreamingUpdateProcessor(with_update_sink(config, store), store, Event(), None)
 
             # KeyboardInterrupt is not an Exception, so the loop cannot catch it.
             def explode(*args, **kwargs):
@@ -599,7 +600,7 @@ def _handle_errors_without_waiting(retry, errors):
     is pre-set so the interruptible wait returns at once."""
     store = InMemoryFeatureStore()
     config = Config(sdk_key='sdk-key', initial_reconnect_delay=1)
-    sp = StreamingUpdateProcessor(config, store, Event(), None, retry_state=retry)
+    sp = StreamingUpdateProcessor(with_update_sink(config, store), store, Event(), None, retry_state=retry)
     sp._running = True
     sp._stop_event.set()
     for error in errors:
@@ -660,7 +661,7 @@ def test_an_extended_regime_wait_is_cut_short_by_stop():
         server.for_path('/all', BasicResponse(401))
         retry = fast_retry_state(ONE_HOUR)
 
-        with StreamingUpdateProcessor(config, store, Event(), None, retry_state=retry) as sp:
+        with StreamingUpdateProcessor(with_update_sink(config, store), store, Event(), None, retry_state=retry) as sp:
             sp.start()
             server.wait_until_request_received()
             # Confirm the wait under test really is long before measuring the stop.
@@ -699,7 +700,7 @@ def test_several_messages_on_one_stream_do_not_extend_the_reset_window():
             # restarted reads back as a different time.
             windows = record_healthy_windows(policy)
             with ticking_clock():
-                with StreamingUpdateProcessor(config, store, ready, None, retry_state=retry) as sp:
+                with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None, retry_state=retry) as sp:
                     sp.start()
                     ready.wait(start_wait)
                     assert sp.initialized()
@@ -733,7 +734,7 @@ def test_a_fresh_stream_starts_a_new_reset_window():
                 )
                 windows = record_healthy_windows(policy)
                 with ticking_clock():
-                    with StreamingUpdateProcessor(config, store, ready, None, retry_state=retry) as sp:
+                    with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None, retry_state=retry) as sp:
                         sp.start()
                         ready.wait(start_wait)
                         assert sp.initialized()
@@ -759,7 +760,7 @@ def test_transport_failures_stay_in_the_normal_regime(error):
     store = InMemoryFeatureStore()
     config = Config(sdk_key='sdk-key', initial_reconnect_delay=1)
     retry = for_streaming(1)
-    sp = StreamingUpdateProcessor(config, store, Event(), None, retry_state=retry)
+    sp = StreamingUpdateProcessor(with_update_sink(config, store), store, Event(), None, retry_state=retry)
     sp._running = True
     sp._stop_event.set()  # so the wait returns at once
 
@@ -775,7 +776,7 @@ def _stream_processor_proxy_test(server, config, secure):
         ready = Event()
         with stream_content(make_put_event()) as stream:
             server.for_path(config.stream_base_uri + '/all', stream)
-            with StreamingUpdateProcessor(config, store, ready, None) as sp:
+            with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None) as sp:
                 sp.start()
                 # Wait till the server has received a request. We need to do this even though do_proxy_tests also
                 # does it, because if we return too soon out of this block, the object returned by stream_content
@@ -799,7 +800,7 @@ def test_records_diagnostic_on_stream_init_success():
             server.for_path('/all', stream)
             diag_accum = _DiagnosticAccumulator(1)
 
-            with StreamingUpdateProcessor(config, store, ready, diag_accum) as sp:
+            with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, diag_accum) as sp:
                 sp.start()
                 ready.wait(start_wait)
                 recorded_inits = diag_accum.create_event_and_reset(0, 0)['streamInits']
@@ -818,7 +819,7 @@ def test_records_diagnostic_on_stream_init_failure():
             server.for_path('/all', error_then_success)
             diag_accum = _DiagnosticAccumulator(1)
 
-            with StreamingUpdateProcessor(config, store, ready, diag_accum) as sp:
+            with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, diag_accum) as sp:
                 sp.start()
                 ready.wait(start_wait)
                 recorded_inits = diag_accum.create_event_and_reset(0, 0)['streamInits']
@@ -845,7 +846,7 @@ def test_status_includes_http_code(status):
             config._data_source_update_sink = DataSourceUpdateSinkImpl(store, listeners, Listeners())
             server.for_path('/all', two_errors_then_success)
 
-            with StreamingUpdateProcessor(config, store, ready, None) as sp:
+            with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None) as sp:
                 sp.start()
                 ready.wait(start_wait)
                 assert sp.initialized()
@@ -886,7 +887,7 @@ def listener(s):
             config._data_source_update_sink = DataSourceUpdateSinkImpl(store, listeners, Listeners())
             server.for_path('/all', SequentialHandler(invalid_stream, valid_stream))
 
-            with StreamingUpdateProcessor(config, store, ready, None) as sp:
+            with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None) as sp:
                 sp.start()
                 ready.wait(start_wait)
                 assert sp.initialized()
@@ -921,7 +922,7 @@ def test_failure_transitions_from_valid():
         config.data_source_update_sink.update_status(DataSourceState.VALID, None)
         server.for_path('/all', error_handler)
 
-        with StreamingUpdateProcessor(config, store, ready, None) as sp:
+        with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None) as sp:
             sp.start()
             server.wait_until_request_received()
             wait_until(lambda: len(spy.statuses) == 2)
@@ -950,7 +951,7 @@ def test_records_environment_id_from_stream_headers():
             config._data_source_update_sink = sink
             server.for_path('/all', stream)
 
-            with StreamingUpdateProcessor(config, store, ready, None) as sp:
+            with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None) as sp:
                 sp.start()
                 ready.wait(start_wait)
                 assert sink.environment_id == 'env-abc-123'
@@ -967,7 +968,7 @@ def test_environment_id_is_none_when_not_provided():
             config._data_source_update_sink = sink
             server.for_path('/all', stream)
 
-            with StreamingUpdateProcessor(config, store, ready, None) as sp:
+            with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None) as sp:
                 sp.start()
                 ready.wait(start_wait)
                 assert sink.environment_id is None
@@ -985,7 +986,7 @@ def test_does_not_record_environment_id_from_error_response_headers():
             config._data_source_update_sink = sink
             server.for_path('/all', error_then_success)
 
-            with StreamingUpdateProcessor(config, store, ready, None) as sp:
+            with StreamingUpdateProcessor(with_update_sink(config, store), store, ready, None) as sp:
                 sp.start()
                 ready.wait(start_wait)
                 assert sink.environment_id is None
@@ -1036,3 +1037,8 @@ def record_stream_init(self, timestamp, duration, failed):
     assert failed is False
     assert duration >= 0
     assert duration < 1000
+
+
+def test_constructor_raises_without_update_sink():
+    with pytest.raises(ValueError, match="data_source_update_sink"):
+        StreamingUpdateProcessor(Config("SDK_KEY"), InMemoryFeatureStore(), Event(), None)
diff --git a/ldclient/testing/mock_async_components.py b/ldclient/testing/mock_async_components.py
index c2654ef1..ac8f9169 100644
--- a/ldclient/testing/mock_async_components.py
+++ b/ldclient/testing/mock_async_components.py
@@ -3,6 +3,8 @@
 """
 
 from ldclient.async_feature_store import AsyncInMemoryFeatureStore
+from ldclient.impl.datasource.async_status import AsyncDataSourceUpdateSinkImpl
+from ldclient.impl.listeners import Listeners
 from ldclient.interfaces import AsyncEventProcessor, AsyncUpdateProcessor
 
 
@@ -80,3 +82,14 @@ def initialized(self) -> bool:
 
     def is_alive(self) -> bool:
         return True
+
+
+def with_async_update_sink(config, store):
+    """
+    Gives the config an async data source update sink that writes to the
+    store, if the config has no sink. The data system does this for a real
+    client. Returns the config.
+    """
+    if config.data_source_update_sink is None:
+        config._data_source_update_sink = AsyncDataSourceUpdateSinkImpl(store, Listeners(), Listeners())
+    return config
diff --git a/ldclient/testing/mock_components.py b/ldclient/testing/mock_components.py
index ad93b32b..6bf3a2ce 100644
--- a/ldclient/testing/mock_components.py
+++ b/ldclient/testing/mock_components.py
@@ -1,6 +1,8 @@
 import time
 from typing import Callable
 
+from ldclient.impl.datasource.status import DataSourceUpdateSinkImpl
+from ldclient.impl.listeners import Listeners
 from ldclient.interfaces import (
     BigSegmentStore,
     BigSegmentStoreMetadata,
@@ -54,3 +56,14 @@ def __init__(self, selector: Selector):
 
     def selector(self) -> Selector:
         return self._selector
+
+
+def with_update_sink(config, store):
+    """
+    Gives the config a data source update sink that writes to the store, if
+    the config has no sink. The data system does this for a real client.
+    Returns the config.
+    """
+    if config.data_source_update_sink is None:
+        config._data_source_update_sink = DataSourceUpdateSinkImpl(store, Listeners(), Listeners())
+    return config
diff --git a/ldclient/testing/test_file_data_source.py b/ldclient/testing/test_file_data_source.py
index b8e3fb0b..b10c378c 100644
--- a/ldclient/testing/test_file_data_source.py
+++ b/ldclient/testing/test_file_data_source.py
@@ -18,6 +18,7 @@
     DataSourceState,
     DataSourceStatus
 )
+from ldclient.testing.mock_components import with_update_sink
 from ldclient.testing.test_util import SpyListener
 from ldclient.versioned_data_kind import FEATURES, SEGMENTS
 
@@ -119,7 +120,7 @@ def teardown_function():
 
 def make_data_source(config, **kwargs):
     global data_source
-    data_source = Files.new_data_source(**kwargs)(config, store, ready)
+    data_source = Files.new_data_source(**kwargs)(with_update_sink(config, store), store, ready)
     return data_source
 
 
@@ -337,3 +338,8 @@ def test_does_not_allow_unsafe_yaml():
         if client is not None:
             client.close()
     assert unsafe_yaml_caused_method_to_be_called is False
+
+
+def test_constructor_raises_without_update_sink():
+    with pytest.raises(ValueError, match="data_source_update_sink"):
+        Files.new_data_source(paths=[])(Config("SDK_KEY"), store, ready)