[OPIK-7186] [SDK] feat: surface batch-flush data loss via FlushResult - #7513
Conversation
The background sender silently dropped trace/span batches on terminal failures (e.g. 403), while flush()/end() reported success. Add in-band, non-blocking, non-throwing visibility: - FlushResult returned by flush()/end() (flushed, drops, pending_replay, per-drop details); flush() bool now reflects data loss. - Opik.last_flush_result + Opik.get_upload_errors() (sender-wide history). - DataLossTracker records terminal drops; FlushReporter assembles results. - Drops recorded across processor (HTTP/serialization/unknown, 401), queue overflow (generic on_evict hook), and shutdown-replay abandonment. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
github-actions
Bot
added
🔴 size/XL
python
⏱️ pre-commit per-hook timing
⏭️ 36 skipped (no matching files changed)
|
![baz-reviewer[bot]](https://avatars.githubusercontent.com/in/933528?s=60&v=4){kind=small}
| if flush and close_on_zero: | ||
| with self._lock: | ||
| entry = self._entries.get(key) |
There was a problem hiding this comment.
Concurrent flushes race on shared bundle
Two racing release() calls can both pass the entry.refcount > 1 check and then call shared_bundle.flush(timeout) concurrently, so the same shared resource gets drained twice and FlushResult can become inconsistent — should we mark the bundle as flush-in-progress under self._lock?
Want Baz to fix this for you? Activate Fixer
Other fix methods
Prompt for AI Agents
Before applying, verify this suggestion against the current code. In
sdks/python/src/opik/api_objects/connection_resources.py around lines 365-375 inside
ConnectionResourceManager.release (the shared pre-flush block where `entry.refcount > 1`
leads to `shared_bundle.flush(timeout)` outside the lock), prevent two racing releases
from both flushing the same shared bundle. Refactor the preflush decision so that, while
holding self._lock, you atomically mark the bundle/entry as “flush in progress”
(e.g., add an `entry.preflush_in_progress` flag and optionally store a flush-result
placeholder) and only the first caller may perform `shared_bundle.flush(timeout)`; any
concurrent callers should not call flush again. If you need consistent FlushReporter
outcomes for all callers, have the non-owner callers either wait for a completion Event
set by the owner (without starting a second flush) or return a clearly defined value
(e.g., flushed=None) while the owner drains and records the authoritative result.
There was a problem hiding this comment.
Skipping — this shared-bundle pre-flush block predates this PR (the change here only captures its return value). A concurrent double streamer.flush() is benign: the drain is idempotent (the second flush finds an already-drained queue), and each client builds its own FlushResult from its own marker, so there's no cross-client inconsistency. Adding lock-held flush-in-progress state is the extra concurrency machinery this MVP-scoped PR deliberately avoids.
🤖 Reply posted via /address-github-pr-comments
There was a problem hiding this comment.
Thanks, that makes sense — I’ll remember that this pre-existing shared-bundle flush path is intentionally tolerated here and save it to memory once the PR is merged.
| bundle = entry.resources | ||
|
|
||
| bundle.close(timeout, flush=flush) | ||
| return bundle.close(timeout, flush=flush) |
There was a problem hiding this comment.
Teardown returns false instead of none
Lease.release() returns False on the flush=False last-reference path because ConnectionResourceManager.release() forwards bundle.close(timeout, flush=flush), so the public API no longer matches its None contract for non-draining teardown — should we normalize that branch to None before returning?
Want Baz to fix this for you? Activate Fixer
Other fix methods
Prompt for AI Agents
Before applying, verify this suggestion against the current code. In
sdks/python/src/opik/api_objects/connection_resources.py around lines 341-401, in
ConnectionResourceManager.release(), the last-reference branch returns
bundle.close(timeout, flush=flush), which yields False for flush=False and violates
Lease.release()’s “None when no draining occurred” contract. Refactor so the
last-reference path captures the close result and, when flush=False, returns None
instead of False; keep the existing behavior for flush=True (return the drained/flush
outcome). Ensure the method’s Optional[bool] semantics now align with
Lease.release’s docstring for non-flushing teardown.
There was a problem hiding this comment.
Fixed in d8ae4a0 — ConnectionResourceManager.release() now returns None on the flush=False path instead of forwarding close()'s bool, matching the documented "no drain outcome" contract.
🤖 Reply posted via /address-github-pr-comments
There was a problem hiding this comment.
Commit d8ae4a0 addressed this comment by changing the last-reference release path to return None when flush=False instead of forwarding bundle.close()'s False result. It preserves the flush outcome only for flush=True, so the non-draining teardown now matches the public None contract.
| def build_result(self, marker: int, *, flushed: bool) -> "data_loss.FlushResult": | ||
| dropped_messages, failures = self._data_loss_tracker.drops_since(marker) | ||
| result = data_loss.FlushResult( | ||
| flushed=flushed, | ||
| remaining_queue_size=self._streamer.queue_size(), | ||
| dropped_messages=dropped_messages, | ||
| dropped_items=sum(failure.item_count for failure in failures), | ||
| failures=failures, | ||
| ) |
There was a problem hiding this comment.
build_result computes dropped_items by summing failure.item_count over the bounded failures window, so once DataLossTracker.drops_since() evicts older entries dropped_items undercounts while dropped_messages stays exact — should we keep a running item-loss accumulator in DataLossTracker (independent of eviction) and use that here instead?
Want Baz to fix this for you? Activate Fixer
Other fix methods
Prompt for AI Agents
In sdks/python/src/opik/message_processing/flush_reporter.py around lines 38-46 in
`FlushReporter.build_result`, stop computing `FlushResult.dropped_items` by summing
`failure.item_count` over the truncated `failures` list returned by
`DataLossTracker.drops_since()`. Instead, update `DataLossTracker` to maintain an exact
cumulative item-loss counter alongside its existing recorded marker/count (so eviction
of the bounded deque cannot affect the total), and expose that total via
`drops_since()`'s return value or a dedicated accessor. Then modify `build_result` to
set `dropped_items` from that exact accumulator, ensuring `dropped_messages` and
`dropped_items` don't diverge after high-error bursts trigger eviction. Update or extend
unit tests to cover the eviction scenario where the old approach previously
undercounted.
There was a problem hiding this comment.
Fixed in d8ae4a0 — DataLossTracker now keeps exact running message and item totals independent of the bounded window, and drops_since() returns an exact dropped_items. It no longer undercounts once details are evicted.
🤖 Reply posted via /address-github-pr-comments
There was a problem hiding this comment.
Commit d8ae4a0 addressed this comment by adding an exact running item-loss total to DataLossTracker that is independent of the bounded deque, so evictions no longer affect dropped_items. FlushReporter.build_result() now uses the tracker’s returned dropped_items directly instead of recomputing it from the truncated failure window.
| """ | ||
| End the Opik session and submit all pending messages. | ||
|
|
There was a problem hiding this comment.
end() doc hides drop mode
end() says it submits all pending messages, which is inaccurate for the public flush=False path that deliberately drops queued data — should we qualify this so only flush=True drains and persists the queue?
Want Baz to fix this for you? Activate Fixer
Other fix methods
Prompt for AI Agents
Before applying, verify this suggestion against the current code. In
sdks/python/src/opik/api_objects/opik_client.py around lines 1884-1896, in the
`Opik.end()` docstring, the first summary sentence currently says it “submit[s] all
pending messages”, which contradicts the `flush: bool = True` behavior and the
described fire-and-forget teardown when `flush=False`. Update that opening sentence to
explicitly state that pending queued messages (including file uploads) are
drained/submitted only when `flush=True`, and when `flush=False` the method does not
wait and may drop anything still in flight. Keep the rest of the docstring consistent
with the existing Args/Returns sections describing `flush=False` as dropping pending
data.
There was a problem hiding this comment.
Commit d8ae4a0 addressed this comment by updating Opik.end()’s docstring to say that pending messages are submitted only when flush=True, and that flush=False drops anything still queued. The surrounding return/behavior docs were also aligned with that distinction.
There was a problem hiding this comment.
Fixed in d8ae4a0 — end()'s docstring now states pending messages are submitted only when flush=True, and dropped when flush=False.
🤖 Reply posted via /address-github-pr-comments
| flushed = self._lease.release(timeout, flush=flush, close_on_zero=True) | ||
| self._finalizer.detach() | ||
| if not flush: | ||
| return None | ||
| self._last_flush_result = self._flush_reporter.build_result( | ||
| marker, flushed=bool(flushed) | ||
| ) | ||
| return self._last_flush_result |
There was a problem hiding this comment.
Idempotent end overwrites success
end() coerces Lease.release() returning None into bool(flushed) and rebuilds _last_flush_result, so a second end() call overwrites the real flush outcome with FlushResult(flushed=False, ...) even though no new shutdown work ran — should we short-circuit when flushed is None and keep the existing last_flush_result?
Want Baz to fix this for you? Activate Fixer
Other fix methods
Prompt for AI Agents
Before applying, verify this suggestion against the current code. In
sdks/python/src/opik/api_objects/opik_client.py around lines 1884-1938 in the `end()`
method, fix idempotency: `self._lease.release(..., flush=flush, ...)` can return `None`
when already released, but the current code still calls
`_flush_reporter.build_result(marker, flushed=bool(flushed))`, overwriting
`_last_flush_result` with a “flushed=False” result on subsequent `end()` calls.
Refactor so that after `flushed = self._lease.release(...)`, if `flush` is True and
`flushed is None`, you short-circuit and return the existing `self._last_flush_result`
(or `None` if it was never set) without rebuilding. Keep the existing behavior for the
first successful release (when `flushed` is not `None`) and still return `None` when
`flush` is False.
There was a problem hiding this comment.
Commit d8ae4a0 addressed this comment by short-circuiting end() when self._lease.release(...) returns None. On repeated end() calls, it now returns the existing self._last_flush_result instead of rebuilding a new FlushResult(flushed=False, ...), preserving the original flush outcome.
There was a problem hiding this comment.
Fixed in d8ae4a0 — end() now short-circuits when release() returns None (already released), keeping the prior last_flush_result instead of overwriting it with a spurious flushed=False result. Added idempotency test coverage.
🤖 Reply posted via /address-github-pr-comments
| Returns: | ||
| True if all messages have been sent within specified timeout, False otherwise. | ||
| True if all messages were delivered within the timeout with no data | ||
| loss; False if the timeout was hit or any message was dropped. |
There was a problem hiding this comment.
FlushResult hides upload failures
flush() now promises True if all messages were delivered ... with no data loss, but it only reflects Streamer.flush() while FileUploadManager.flush() still returns True after failed uploads, so Opik.flush() and last_flush_result.success can report success even when attachment uploads were lost — should we scope the public contract to trace/span delivery only, or include upload failures in the flush outcome?
Want Baz to fix this for you? Activate Fixer
Other fix methods
Prompt for AI Agents
Before applying, verify this suggestion against the current code. In
sdks/python/src/opik/api_objects/opik_client.py around lines 1940-1975, in the `flush()`
method (and the `last_flush_result` property that it feeds), fix the breaking contract
mismatch: today `flush()` computes `last_flush_result` solely from
`self._streamer.flush(timeout)`, but file/attachment uploads can fail independently
(FileUploadManager.flush() can return success even with per-upload failures). Update
`flush()` so the public `last_flush_result.success` and docstring reflect the true
outcome by either (preferred) aggregating streamer message delivery success plus
file-upload failure/partial-loss into the boolean passed to
`_flush_reporter.build_result`, or (alternative) narrowing the doc/result wording to
“trace/span delivery” only and making `last_flush_result` name/semantics match that
narrower scope. Ensure the same semantics are consistent with `end()` (lines ~1884-1939)
and that the reported success includes (or explicitly excludes) attachment upload
failures.
There was a problem hiding this comment.
Commit d8ae4a0 addressed this comment by narrowing flush()/last_flush_result to trace/span/feedback delivery and explicitly stating that attachment/file uploads are not reflected in the outcome. It also made end() return the same flush outcome path, so the public semantics are now consistent with that narrower contract.
There was a problem hiding this comment.
Fixed in d8ae4a0 — flush()'s docstring now scopes the contract to trace/span/feedback delivery and notes attachment/file uploads aren't reflected in the outcome. (Upload-failure accounting is deferred to a follow-up per this PR's MVP scope.)
🤖 Reply posted via /address-github-pr-comments
…t, exact dropped_items - end(): short-circuit when release() returns None so a repeated end() no longer overwrites last_flush_result with a spurious not-flushed result - flush(): on failure, record a failed FlushResult instead of leaving a stale prior success in last_flush_result - manager.release(): return None (not close()'s bool) on the flush=False path, matching the "no drain outcome" contract - DataLossTracker: keep exact running message/item totals so dropped_items no longer undercounts once details are evicted from the bounded window - revert recording unparseable-429s as terminal data loss (a 429 is transient, not lost) — restores prior behavior - docs: qualify end()/flush() for flush=False and attachment-upload scope - test: build the flush-result client via __new__ to avoid real background resources; add idempotent-end and stale-result coverage Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
| self._data_loss_tracker.record( | ||
| data_loss.FailedMessageInfo( | ||
| message_type=type(message).__name__, | ||
| reason=reason, |
There was a problem hiding this comment.
❓ question | Logic
Every terminal branch now records a drop except one: a 429 whose exception.headers is None or whose rate_limit.parse_rate_limit(...) returns None (a few lines up in the elif exception.status_code == 429: block). In that case the handler neither re-raises OpikCloudRequestsRateLimited (so it isn't retried/re-enqueued) nor calls _record_data_loss, so the message is silently abandoned and stays invisible to FlushResult.
Given the PR's goal of surfacing terminal drops in-band, is leaving this 429 sub-case unrecorded intentional? It's pre-existing behavior, but this change makes it the one terminal path that escapes accounting. If it's out of scope, a one-line note in the ## Details "deferred" list would make the boundary explicit.
🤖 Review posted via /review-github-pr
There was a problem hiding this comment.
Good catch — it's intentional (the 429 branch is byte-identical to main), but you're right it should be explicit. Added a note to the PR description's deferred list: the header-less / unparseable 429 sub-case is neither retried nor recorded, left as-is to avoid changing rate-limit handling in this PR.
| """ | ||
| Flush the streamer to ensure all messages are sent. | ||
|
|
||
| Covers delivery of trace/span/feedback messages; attachment/file uploads |
There was a problem hiding this comment.
🧹 nit | Docs
attachment/file uploads are not reflected in the outcome
This is only true for the data-loss detail (failures / dropped_*). The flushed field — and therefore success and this method's return value — does reflect uploads: streamer.flush() returns upload_flushed and self._all_done(), so a stuck/failed upload makes flushed=False and this returns False.
Consider tightening to something like: "upload failures are not counted in the dropped-message detail, though an incomplete upload still makes the flush report as not fully flushed."
🤖 Review posted via /review-github-pr
There was a problem hiding this comment.
Commit 86e3b46 addressed this comment by documenting that flush covers attachment/file uploads while distinguishing the flush outcome from its data-loss detail.
There was a problem hiding this comment.
Fixed — you're right, flushed does reflect uploads (streamer.flush() returns upload_flushed and _all_done()). Reworded the docstring: attachment/file upload failures aren't counted in the data-loss detail (dropped_*/failures), but an incomplete upload still makes the flush report as not fully flushed, so the returned bool does reflect uploads.
| detail: Optional[str] = None | ||
| timestamp: float = dataclasses.field(default_factory=time.time) |
There was a problem hiding this comment.
🧹 nit | Style
FlushResult is frozen=True but carries a mutable failures: List[FailedMessageInfo], so the value isn't truly immutable — a caller can still result.failures.append(...). It's never mutated internally, so this is cosmetic, but a Tuple[FailedMessageInfo, ...] would make the frozen contract honest (and drops_since already builds a fresh list you can just wrap).
🤖 Review posted via /review-github-pr
There was a problem hiding this comment.
Commit ef23478 addressed this comment by changing both failures fields to tuples and converting the constructed reports with tuple(failures), preventing callers from appending to the returned collection.
There was a problem hiding this comment.
Done — FlushResult.failures (and ErrorsReport.failures) are now Tuple[FailedMessageInfo, ...], built via tuple(...), so the frozen contract is honest.
|
👋 Review summary What looks good
Overall Inline comments: 1 question + 2 nits — nothing blocking. 🤖 Review posted via /review-github-pr |
| @@ -171,6 +188,14 @@ def process(self, message: messages.BaseMessage) -> None: | |||
| extra={"error_tracking_extra": error_tracking_extra}, | |||
There was a problem hiding this comment.
❓ question | Semantics (platform-side, non-blocking — likely a follow-up)
The ## Details motivating example is "HTTP 403 on an exhausted trace pool." Tracing the managed-cloud path, the actual rejection for an exhausted allocation is the free-tier OPIK_SPAN_COUNT quota, enforced on the @UsageLimited trace/span write endpoints, which throws 402 Payment Required with "You have exceeded the usage limit for this operation." — not 403 (403 is used for genuine authorization failures elsewhere).
This doesn't affect the code here: both 402 and 403 map to HTTP_CLIENT_ERROR → terminal drop, so the accounting is correct regardless. Two small things worth considering, both fine as a follow-up:
- PR-description wording — the example is really a 402 free-tier span-quota rejection, not a 403; tightening the wording keeps the user-facing story accurate ("you've hit your span allowance, upgrade/reduce volume" vs. "forbidden").
- Should quota-exceeded be a distinct
FailureReason? A genericHTTP_CLIENT_ERRORis technically right but loses the actionable meaning. A dedicated reason (e.g.USAGE_LIMIT_EXCEEDEDderived from 402) would let callers surface "upgrade your plan" rather than a generic client error. Out of this PR's scope, but a natural extension of the newFailureReasontaxonomy.
🤖 Review posted via /review-github-pr
There was a problem hiding this comment.
Thanks for tracing the managed-cloud path. Tightened the PR description to "an HTTP 4xx ... a 402 free-tier span-quota rejection, or a 403" so the user-facing story is accurate. A dedicated USAGE_LIMIT_EXCEEDED reason derived from 402 is a nice extension of the new FailureReason taxonomy, but out of scope here — both 402 and 403 map to HTTP_CLIENT_ERROR today so accounting is already correct. Noting it as a follow-up.
JetoPistola
left a comment
JetoPistola
left a comment
There was a problem hiding this comment.
Approving — solid, well-scoped implementation with strong test coverage. The one open item (402-vs-403 status semantics / a dedicated USAGE_LIMIT_EXCEEDED FailureReason) is platform-side and non-blocking; good candidate for a follow-up. Nice work.
🤖 Review posted via /review-github-pr
last_flush_result is scoped to a single flush window, so drops that happened before or between flushes were recorded but not retrievable in-band. Expose the DataLossTracker's retained history via Opik.get_data_loss() (tracker -> FlushReporter -> client). Storage stays a capped deque, so memory is bounded regardless of run length. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…eport() Replace the bare-list get_data_loss() with get_errors_report() returning an ErrorsReport instance: exact all-time dropped message/item totals, retained per-drop details, a generated_at timestamp, and first/last_failure_at helpers (each FailedMessageInfo already carries its own timestamp). Storage stays a capped deque, so memory is bounded. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Callers can check total_dropped_messages directly; the convenience flag is redundant. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Make the bounded (drop-oldest, default 1000) nature of the report's per-drop details explicit on ErrorsReport and Opik.get_errors_report(): totals stay exact, but failures keeps only the most recent entries. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
petrotiurin
left a comment
petrotiurin
left a comment
There was a problem hiding this comment.
Other than this looks good, no issues from my side.
Per review: the deque is thread-safe and an exact-to-the-message data-loss tally isn't worth lock contention on the hot sending path. Counters are plain ints now (best-effort under concurrent drops); reads snapshot the deque once. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Address PR review: frozen dataclasses now expose failures as Tuple[FailedMessageInfo, ...] so the value is truly immutable; clarify that flush()'s bool does reflect upload completion (only the data-loss detail excludes uploads); strengthen the eviction test to assert retained identities and order. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…h-error-visibility' into aliaksandrk/OPIK-7186-batch-flush-error-visibility
3 participants
Details
Opik.end()/flush()silently swallowed batch POST failures (e.g. an HTTP 4xx on an exhausted trace pool — a402free-tier span-quota rejection, or a403): the batch error was only logged, never surfaced via a return value or exception, so caller code saw success while trace/span data was dropped server-side. This PR surfaces terminal data loss in-band — without ever raising into user code or blocking the app, since an observability SDK must not disrupt its host.What it adds
FlushResult(exported fromopik) — returned byflush()/end()and cached onOpik.last_flush_result. Fields:flushed,remaining_queue_size,dropped_messages,dropped_items,failures(list ofFailedMessageInfo), and a.successproperty (flushed and dropped_messages == 0).FailedMessageInfo/FailureReason(exported) — per-drop detail: message type, item count, HTTP status, reason (HTTP_CLIENT_ERROR,HTTP_SERVER_ERROR,UNAUTHORIZED,SERIALIZATION,UNKNOWN), and a short detail string.Opik.get_errors_report() -> ErrorsReport(exported) — sender-wide data-loss report, independent of any single flush: surfaces drops that happened before or between flushes, whichlast_flush_result(scoped to one flush window) does not. Carries exact all-timetotal_dropped_messages/total_dropped_items, retained per-dropfailures(each with its owntimestamp), agenerated_attimestamp, andfirst_failure_at/last_failure_at/has_data_losshelpers. Backed by a capped collection (bounded deque, most-recent N), so memory stays bounded regardless of run length.DataLossTracker— thread-safe, bounded record of terminal drops shared acrossOpikhandles on one connection; keeps exact running message/item totals independent of the bounded detail window, so counts stay correct even after eviction.FlushReporter— assembles aFlushResultfrom the streamer + tracker; owns the "did this flush lose data" logic so the client doesn't gather it piecemeal.BaseMessage.item_count— how many traces/spans a (possibly batched) message carries.What it affects (behavior / API changes)
flush() -> boolnow means "delivered everything": it returnsFalse(never raises) on a timeout or any terminal drop — previously it only reflected whether the queue drained. Callers that checkif client.flush(): ...will now correctly seeFalsewhen data was lost.end() -> Optional[FlushResult](wasNone) — returns the flush outcome, derived from the authoritative flush result computed during teardown (not the weakerqueue_size() == 0proxy). Idempotent: a repeatedend()no longer overwriteslast_flush_resultwith a spurious not-flushed result.Streamer.close()now returns whether all data was flushed (authoritative onflush=True); threaded throughSharedConnectionResourcesBundle.close()/flush()and the lease/manager release path.HTTP_CLIENT_ERROR, 5xx / exhausted retries→HTTP_SERVER_ERROR, 401→UNAUTHORIZED, serialization→SERIALIZATION, unexpected→UNKNOWN. Transient states are deliberately not counted as loss: 409 dedup, retryable 429, and connection errors parked for replay.OpikMessageProcessorand the processors-chain factory take adata_loss_tracker; the bundle builds the tracker +FlushReporter. No change to transient-error handling.Scope / deferred
Scope is intentionally the reported failure mode: backend rejection of trace/span batches surfaced through the flush contract. Deferred to follow-ups: queue-overflow accounting, shutdown-replay-abandoned accounting, file-upload (attachment) failure counts, and the header-less/unparseable
429sub-case (a pre-existing path that is neither retried nor recorded — left as-is to avoid changing rate-limit handling in this PR).Change checklist
Issues
AI-WATERMARK
AI-WATERMARK: yes
Testing
pytest tests/unit/message_processing/ tests/unit/api_objects/test_opik_client.py tests/unit/api_objects/test_connection_resource_manager.py— all green.pre-commit(ruff, ruff-format, mypy) clean on changed files.flush()/end()return values andlast_flush_result; never-raises on internal failure (and no stale prior success); authoritativeflushed(notqueue_size() == 0); idempotent repeatedend(); per-branch terminal-drop recording (403/500/401/serialization/retry) vs non-recording of transient states (409/429/connection);DataLossTrackerexact message/item counts under eviction;get_errors_report()surfaces drops outside the flush window, carries timestamps, and stays bounded to capacity.success=True; a real backend rejection (invalid key → 401) yieldsflush() == Falsewith populatedlast_flush_resultand no exception.Documentation
No external documentation changes required. Public docstrings for
end(),flush(),last_flush_result, and the newFlushResult/FailedMessageInfo/FailureReasontypes are added inline.