Most Airbyte custom-connector tutorials assume the happy path: the upstream API always responds, pagination is clean, and a sync either fully succeeds or fully fails. Production APIs don't work that way. They rate-limit you mid-page, drop connections after 30 seconds, and occasionally return a 200 with a truncated JSON body. If your connector's incremental sync logic can't recover from that without re-pulling the entire stream, you're paying for redundant API calls and burning your rate limit budget on data you already have.
This walkthrough builds a Python CDK HttpStream subclass that treats partial failure as the normal case, not the exception: it checkpoints state inside a sync (not just at the end), resumes from the last confirmed cursor instead of the start of the stream, and has a test harness that actually simulates the API being unreliable.
The failure mode default connectors don't handle
Airbyte's incremental sync model is built around a cursor field — usually a timestamp or auto-incrementing ID — and a state message the platform persists between syncs. The naive implementation reads stream_state, requests everything after that cursor, and emits a new state message once the entire stream finishes. That's fine until page 40 of 60 times out. Without intermediate checkpoints, the next sync attempt restarts at page 1, because the state message was never emitted.
For a stream with 500K records paginated at 100 per page, that's the difference between resuming 40 pages in and re-fetching 4,000 pages you already had.
Checkpointing inside the slice, not just after it
The CDK exposes state_checkpoint_interval on HttpStream, which tells the platform to emit a state message every N records instead of only at stream completion. This is the single highest-leverage change you can make to an existing connector:
class Orders(HttpStream, IncrementalMixin):
cursor_field = "updated_at"
primary_key = "id"
state_checkpoint_interval = 500
def __init__(self, start_date: str, **kwargs):
super().__init__(**kwargs)
self._cursor_value = start_date
@property
def state(self):
return {self.cursor_field: self._cursor_value}
@state.setter
def state(self, value):
self._cursor_value = value.get(self.cursor_field, self._cursor_value)
def request_params(self, stream_state, **kwargs):
cursor = (stream_state or {}).get(self.cursor_field, self._cursor_value)
return {"updated_since": cursor, "sort": "updated_at:asc", "limit": 100}
def parse_response(self, response, **kwargs):
records = response.json().get("data", [])
for record in records:
if record[self.cursor_field] > self._cursor_value:
self._cursor_value = record[self.cursor_field]
yield record
The critical detail most tutorials skip: sorting updated_since ascending by cursor field is a hard requirement, not an optimization. If the API returns records out of cursor order, state_checkpoint_interval will checkpoint a cursor value that skips records emitted later in the same page — a silent data loss bug that won't show up until someone notices a gap in a downstream dashboard weeks later. If the upstream API can't guarantee sort order, don't use interval checkpointing; checkpoint only after you've buffered and sorted a full page client-side.
Resuming mid-pagination after a timeout
Checkpointing state doesn't help if your next_page_token logic can't resume from an arbitrary cursor value — it needs to resume from wherever the last successful checkpoint left off, which may be mid-page from the API's pagination perspective. The cleanest way to handle this is to make pagination cursor-driven rather than page-number-driven:
def next_page_token(self, response):
records = response.json().get("data", [])
if len(records) < 100:
return None
return {"updated_since": records[-1][self.cursor_field]}
def request_params(self, stream_state, next_page_token=None, **kwargs):
cursor = (next_page_token or {}).get("updated_since") \
or (stream_state or {}).get(self.cursor_field, self._cursor_value)
return {"updated_since": cursor, "sort": "updated_at:asc", "limit": 100}
With page-number pagination, a timeout on page 40 means you must know you were on page 40 to resume correctly — and page numbers don't survive a sync restart cleanly if state_checkpoint_interval fired mid-page. With cursor-driven pagination, the next request is self-describing: whatever cursor value was last checkpointed is exactly where the next request should start, regardless of which page that corresponded to in the failed attempt.
Backoff that respects Airbyte's own retry budget
The CDK's default should_retry and backoff_time handle standard 429/5xx responses, but they retry within a single attempt's lifetime — they won't help if the upstream API rate-limits you for minutes at a time, longer than a single sync's retry budget makes sense to burn synchronously. For APIs with long rate-limit windows, read the Retry-After header and cap how long you'll wait in-process versus letting the sync attempt fail cleanly so Airbyte's scheduler retries it later:
def backoff_time(self, response):
retry_after = response.headers.get("Retry-After")
if retry_after:
wait = float(retry_after)
if wait > 120:
raise AirbyteTracedException(
message=f"Rate limited for {wait}s, exceeds in-sync retry budget",
failure_type=FailureType.transient_error,
)
return wait
return None
Raising AirbyteTracedException with failure_type=transient_error matters here: it tells Airbyte's UI and any downstream alerting that this failure is expected to resolve on its own, not a connector bug that needs a code fix. Teams that skip this distinction end up with on-call rotations investigating rate-limit backoffs as if they were broken connectors.
Testing failure, not just success
A connector's happy-path unit tests tell you almost nothing about whether it survives production. Build a lightweight mock server (using responses or pytest-httpserver) that deliberately misbehaves on specific requests:
def test_resumes_from_checkpoint_after_timeout(mock_server):
mock_server.expect_request("/orders", query_string="updated_since=2026-01-01") \
.respond_with_json({"data": FIRST_PAGE})
mock_server.expect_request("/orders", query_string=f"updated_since={FIRST_PAGE[-1]['updated_at']}") \
.respond_with_handler(lambda r: (_ for _ in ()).throw(Timeout()))
stream = Orders(start_date="2026-01-01", url_base=mock_server.url_for(""))
records = list(itertools.islice(stream.read_records(sync_mode=SyncMode.incremental), len(FIRST_PAGE)))
assert stream.state["updated_at"] == FIRST_PAGE[-1]["updated_at"]
This test asserts something the happy-path suite never checks: that after a mid-stream timeout, the connector's state property reflects exactly the last successfully processed record — not the start of the sync, not a stale value from the previous checkpoint interval. Run this pattern for each failure mode you expect in production: connection timeout, truncated JSON body, 429 with and without Retry-After, and a 500 on the very first page (state should remain untouched).
The checklist
Before shipping a custom connector against a production API, verify: cursor field sort order is guaranteed by the API or enforced client-side; pagination resumes from cursor value, not page number; state_checkpoint_interval is set to a value that bounds re-fetch cost without checkpointing so often it adds overhead; rate-limit responses distinguish transient_error from connector bugs; and at least one test asserts state correctness after a simulated failure, not just record correctness after a simulated success. None of this shows up in a quickstart, and all of it is what separates a connector that works in a demo from one that survives a flaky API in production.
Top comments (0)