feat: continuous polling loop for Schiphol producer #1

New Issue

peter · 2026-07-19T08:39:50+02:00

peter commented

2026-07-19 08:39:50 +02:00

Problem Statement

The producer currently exits after completing a single Poll — it fetches all pages from the Schiphol API, publishes one Snapshot per Flight to Kafka, and stops. This means the Kafka topic goes stale immediately after the first run. For consumers (such as the planned database writer) to always have current flight data, the producer must continuously re-poll and keep the topic up to date.

Solution

Refactor the producer so it runs as a long-lived process: it polls the Schiphol API once, publishes Snapshots for all Flights, waits 5 minutes, then polls again — indefinitely. The HTTP session, Kafka engine, and OpenTelemetry tracer are initialised once at startup and reused across Polls.

User Stories

As an operator, I want the producer to run continuously without manual re-invocation, so that the Kafka topic always reflects current flight data.
As an operator, I want each Poll to complete a full sweep of all paginated Schiphol API pages, so that no Flights are missed.
As an operator, I want the producer to wait 5 minutes between Polls, so that the API is not hammered unnecessarily.
As an operator, I want the producer to log how many Flights were published after each Poll, so that I can verify it is working correctly.
As an operator, I want the producer to continue running if a single Poll fails (e.g. transient API error), so that one failure doesn't take down the whole pipeline.
As an operator, I want the producer to shut down gracefully when a SIGTERM/SIGINT is received, so that in-flight Kafka sends are not corrupted.
As an operator, I want the OTel tracer to emit a span per Poll (containing the Flight count), so that I can track polling health in my observability stack.
As a developer, I want the poll logic isolated in a single function, so that I can test it independently of the startup and teardown concerns.
As a developer, I want the poll function to accept the HTTP session, Kafka engine, settings, and tracer as arguments, so that I can inject test doubles without patching globals.
As a developer, I want the poll function to return the count of Flights published, so that tests can assert on it without inspecting Kafka.
As a developer, I want errors raised inside a Poll to propagate up to the loop, so that the loop can decide whether to retry or abort.
As a consumer developer, I want each Kafka message to be a full Flight Snapshot keyed by flight ID, so that I can always reconstruct the latest state of any Flight from the topic.

Implementation Decisions

Extract the inner fetch-and-publish sweep from start() into a standalone async poll() function. poll() accepts the HTTP session, Kafka stream engine, API settings, Kafka settings, and tracer as parameters and returns the number of Flights published.
start() is refactored to: (1) initialise OTel, create Kafka engine, create HTTP session; (2) loop forever — call poll(), log the result, sleep 5 minutes, repeat.
The 5-minute interval is a constant (not yet a configurable setting, per current scope).
On transient Poll failure (non-fatal exceptions such as HTTP errors), log the error and continue to the next Poll cycle rather than crashing the process.
Graceful shutdown is handled by the existing aiorun shutdown callback; no changes needed there.
The OTel span wrapping a full Poll (not individual page fetches) is a new span — poll_cycle — carrying a flights.published attribute with the count.
The Kafka message format is unchanged: raw JSON bytes of the Schiphol API flight object, keyed by flight ID string. This is consistent with ADR-0001 (snapshots, not events).

Testing Decisions

A good test exercises the observable output of poll() — the Kafka messages sent and the return value — without asserting on internal implementation details like pagination cursor management or span creation.

What to test (at the poll() seam):

Given a mock HTTP session that returns a single page of N flights with no link: rel="next" header, poll() publishes N messages to Kafka and returns N.
Given a mock HTTP session that returns two pages (first page has link: rel="next"), poll() publishes the combined flight count across both pages.
Given a mock HTTP session that raises an HTTP error, poll() propagates the exception.

Prior art: None — this will be the first test in the codebase. Use pytest with pytest-asyncio for async test support. Mock the HTTP session with a simple stub class (not unittest.mock) that returns pre-built response objects. Mock the Kafka engine's send() method to capture calls.

Out of Scope

Making the poll interval configurable via environment variable (a future settings change).
Implementing the consumer that writes Snapshots to a database (separate feature, database not yet decided).
Schema Registry integration on the producer side (the current producer publishes raw JSON; schema enforcement is a separate decision).
Adaptive polling (e.g. backing off when the API is slow).
Metrics beyond OTel tracing (Prometheus, etc.).

Further Notes

ADR-0001 records that Kafka messages are Snapshots, not immutable events. ADR-0002 records the decision to use a continuous process rather than a cron job. Both are relevant background for this feature.

The middelware.py file (note: typo in filename) contains consumer-side deserialization middleware. It is not touched by this feature.

## Problem Statement The producer currently exits after completing a single Poll — it fetches all pages from the Schiphol API, publishes one Snapshot per Flight to Kafka, and stops. This means the Kafka topic goes stale immediately after the first run. For consumers (such as the planned database writer) to always have current flight data, the producer must continuously re-poll and keep the topic up to date. ## Solution Refactor the producer so it runs as a long-lived process: it polls the Schiphol API once, publishes Snapshots for all Flights, waits 5 minutes, then polls again — indefinitely. The HTTP session, Kafka engine, and OpenTelemetry tracer are initialised once at startup and reused across Polls. ## User Stories 1. As an operator, I want the producer to run continuously without manual re-invocation, so that the Kafka topic always reflects current flight data. 2. As an operator, I want each Poll to complete a full sweep of all paginated Schiphol API pages, so that no Flights are missed. 3. As an operator, I want the producer to wait 5 minutes between Polls, so that the API is not hammered unnecessarily. 4. As an operator, I want the producer to log how many Flights were published after each Poll, so that I can verify it is working correctly. 5. As an operator, I want the producer to continue running if a single Poll fails (e.g. transient API error), so that one failure doesn't take down the whole pipeline. 6. As an operator, I want the producer to shut down gracefully when a SIGTERM/SIGINT is received, so that in-flight Kafka sends are not corrupted. 7. As an operator, I want the OTel tracer to emit a span per Poll (containing the Flight count), so that I can track polling health in my observability stack. 8. As a developer, I want the poll logic isolated in a single function, so that I can test it independently of the startup and teardown concerns. 9. As a developer, I want the poll function to accept the HTTP session, Kafka engine, settings, and tracer as arguments, so that I can inject test doubles without patching globals. 10. As a developer, I want the poll function to return the count of Flights published, so that tests can assert on it without inspecting Kafka. 11. As a developer, I want errors raised inside a Poll to propagate up to the loop, so that the loop can decide whether to retry or abort. 12. As a consumer developer, I want each Kafka message to be a full Flight Snapshot keyed by flight ID, so that I can always reconstruct the latest state of any Flight from the topic. ## Implementation Decisions - Extract the inner fetch-and-publish sweep from `start()` into a standalone async `poll()` function. `poll()` accepts the HTTP session, Kafka stream engine, API settings, Kafka settings, and tracer as parameters and returns the number of Flights published. - `start()` is refactored to: (1) initialise OTel, create Kafka engine, create HTTP session; (2) loop forever — call `poll()`, log the result, sleep 5 minutes, repeat. - The 5-minute interval is a constant (not yet a configurable setting, per current scope). - On transient Poll failure (non-fatal exceptions such as HTTP errors), log the error and continue to the next Poll cycle rather than crashing the process. - Graceful shutdown is handled by the existing `aiorun` shutdown callback; no changes needed there. - The OTel span wrapping a full Poll (not individual page fetches) is a new span — `poll_cycle` — carrying a `flights.published` attribute with the count. - The Kafka message format is unchanged: raw JSON bytes of the Schiphol API flight object, keyed by flight ID string. This is consistent with ADR-0001 (snapshots, not events). ## Testing Decisions A good test exercises the observable output of `poll()` — the Kafka messages sent and the return value — without asserting on internal implementation details like pagination cursor management or span creation. **What to test (at the `poll()` seam):** - Given a mock HTTP session that returns a single page of N flights with no `link: rel="next"` header, `poll()` publishes N messages to Kafka and returns N. - Given a mock HTTP session that returns two pages (first page has `link: rel="next"`), `poll()` publishes the combined flight count across both pages. - Given a mock HTTP session that raises an HTTP error, `poll()` propagates the exception. **Prior art:** None — this will be the first test in the codebase. Use `pytest` with `pytest-asyncio` for async test support. Mock the HTTP session with a simple stub class (not `unittest.mock`) that returns pre-built response objects. Mock the Kafka engine's `send()` method to capture calls. ## Out of Scope - Making the poll interval configurable via environment variable (a future settings change). - Implementing the consumer that writes Snapshots to a database (separate feature, database not yet decided). - Schema Registry integration on the producer side (the current producer publishes raw JSON; schema enforcement is a separate decision). - Adaptive polling (e.g. backing off when the API is slow). - Metrics beyond OTel tracing (Prometheus, etc.). ## Further Notes ADR-0001 records that Kafka messages are Snapshots, not immutable events. ADR-0002 records the decision to use a continuous process rather than a cron job. Both are relevant background for this feature. The `middelware.py` file (note: typo in filename) contains consumer-side deserialization middleware. It is not touched by this feature.

peter added the

ready-for-agent

label 2026-07-19 08:39:50 +02:00

peter referenced this issue

2026-07-19 08:45:56 +02:00

prefactor: extract poll() function from start() #3

peter referenced this issue

2026-07-19 08:46:10 +02:00

test: add pytest infrastructure and poll() tests #4

peter referenced this issue

2026-07-19 08:46:25 +02:00

feat: implement continuous polling loop #5