Documentation
¶
Index ¶
Constants ¶
const ( DefaultRetryMaxAttempts = 4 DefaultRetryInitialBackoff = 100 * time.Millisecond DefaultRetryMaxBackoff = 2 * time.Second DefaultRetryDeadline = 10 * time.Second )
Retry defaults. Deliberately short: the point is to absorb a hiccup, not to wait out an outage. A destination that is still refusing after a few seconds is better reported, because the exit code says "retryable" and a supervisor restarts the pipeline anyway.
Retrying is on by default. Before this, one refused connection killed the process, which cost a cold start, a group rejoin and a rebalance to recover from a blip. That default was the defect, not a safe baseline.
Variables ¶
This section is empty.
Functions ¶
func Kinds ¶ added in v1.1.0
func Kinds() []string
Kinds lists every sink type the engine can build, sorted.
func New ¶
func New(ctx context.Context, sink config.Sink, conn adbc.Connection, opts ...Option) (core.Sink, error)
New builds a sink and wraps it in a retry ladder where one helps.
Not every sink is wrapped. The Kafka sink hands records to franz-go, which already retries a produce with its own backoff; a second ladder on top of that one delays the report without improving delivery. The console, noop and sqlcommand sinks reach nothing that can be temporarily unavailable -- the sqlcommand sink writes through the pipeline's own DuckDB connection, and a failure there is not a network blip.
That leaves the sinks that cross a network to somebody else's server.
The context is required rather than optional. It bounds the probe below, which dials, and a convenience overload that supplied context.Background() was how two of the three call sites came to start a pipeline that could hang forever against a host dropping packets rather than refusing them.
Types ¶
type ClickhouseSink ¶
type ClickhouseSink struct {
// contains filtered or unexported fields
}
ClickhouseSink inserts result batches into a ClickHouse table.
The Python sink hands the Arrow table to clickhouse_connect's insert_arrow, which maps Arrow columns to table columns by name. clickhouse-go has no Arrow entry point, so the batch is unpacked into rows against an explicit column list taken from the Arrow schema, which preserves that name-based mapping.
func NewClickhouseSink ¶
func NewClickhouseSink(conf config.ClickhouseSink) (*ClickhouseSink, error)
func (*ClickhouseSink) BufferedRows ¶ added in v1.1.0
func (s *ClickhouseSink) BufferedRows() int
BufferedRows reports the rows this sink is holding that no flush has delivered. The pipeline publishes it as sink_buffered_rows, so an operator can tell a sink retrying a destination from one that has stopped draining.
func (*ClickhouseSink) Close ¶
func (s *ClickhouseSink) Close() error
func (*ClickhouseSink) Flush ¶
func (s *ClickhouseSink) Flush(ctx context.Context) error
Flush delivers the buffered batches, and keeps them buffered if it cannot.
The retry ladder calls Flush again after a retryable failure. Discarding the buffer on the way in made that second call find nothing to send and return nil, so the ladder reported success for rows that never left the process and the pipeline committed their offsets. Batches are released only once ClickHouse has acknowledged them.
func (*ClickhouseSink) Probe ¶ added in v1.0.5
func (s *ClickhouseSink) Probe(ctx context.Context) error
Probe dials the server. clickhouse.Open only validates options, so this is the first time the pipeline learns whether the destination is there.
func (*ClickhouseSink) WriteTable ¶
type ConsoleSink ¶
type ConsoleSink struct {
// contains filtered or unexported fields
}
ConsoleSink writes each result row to stdout as a JSON object.
The rows are held here rather than in a bufio.Writer. bufio latches its first error permanently: once a write failed, every later Flush returned that same stale error and the bytes were gone, so the sink never recovered even after the condition cleared. stdout is not immune to that -- redirect it to a full disk or a pipe whose reader exits and the write fails -- and a sink that cannot recover turns a transient failure into a dead pipeline.
func NewConsoleSink ¶
func NewConsoleSink() *ConsoleSink
func NewConsoleSinkTo ¶
func NewConsoleSinkTo(w io.Writer) *ConsoleSink
func (*ConsoleSink) BufferedRows ¶ added in v1.1.0
func (s *ConsoleSink) BufferedRows() int
BufferedRows reports the rows this sink is holding that no flush has written. Every row ends in a newline and a short write only removes a prefix, so counting newlines counts the rows still owed, including a row left half written.
func (*ConsoleSink) Flush ¶
func (s *ConsoleSink) Flush(ctx context.Context) error
Flush writes the buffered rows, and keeps whatever it could not write.
A short write leaves exactly the unwritten tail pending, so a retry sends the remainder rather than the whole batch again: the reader must not see a row twice because the tail of the batch failed.
func (*ConsoleSink) WriteTable ¶
WriteTable buffers the rows. Nothing reaches the writer until Flush, so the pipeline never commits offsets for a row that only got as far as this process.
type IcebergSink ¶
type IcebergSink struct {
// contains filtered or unexported fields
}
IcebergSink appends result batches to an Iceberg table.
func NewIcebergSink ¶
func NewIcebergSink(ctx context.Context, catalogName, tableName string) (*IcebergSink, error)
func (*IcebergSink) BufferedRows ¶ added in v1.1.0
func (s *IcebergSink) BufferedRows() int
BufferedRows reports the rows this sink is holding that no flush has delivered. The pipeline publishes it as sink_buffered_rows, so an operator can tell a sink retrying a destination from one that has stopped draining.
func (*IcebergSink) Flush ¶
func (s *IcebergSink) Flush(ctx context.Context) error
Flush appends the buffered batches, and keeps whatever it could not append.
The retry ladder calls Flush again after a retryable failure. Discarding the buffer on the way in made that second call find nothing to append and return nil, so the ladder reported success for rows that were never written and the pipeline committed their offsets.
Each batch is a separate append, so a failure part-way through has already committed the batches before it. Only the ones still undelivered are kept; requeueing all of them would append the earlier ones twice.
func (*IcebergSink) WriteTable ¶
type KafkaSink ¶
type KafkaSink struct {
// contains filtered or unexported fields
}
KafkaSink produces one message per result row, JSON encoded, matching the Python KafkaSink.
func NewKafkaSink ¶
NewKafkaSink builds the sink. Extra client options are appended last, so a caller can override what this function set: the conformance test dials every broker address through a proxy, which is the only way to fault the connection from outside. A testcontainers broker advertises its own mapped host port, so a client that merely bootstraps through a proxy reconnects around it on the first metadata response.
func (*KafkaSink) BufferedRows ¶ added in v1.1.0
BufferedRows reports the rows this sink is holding that no flush has had acknowledged. The pipeline publishes it as sink_buffered_rows, so an operator can tell a sink retrying a broker from one that has stopped draining.
func (*KafkaSink) Flush ¶
Flush blocks until every buffered record has been acknowledged, so a batch is durable before its source offsets are committed.
It waits on ctx, not on a background context. franz-go retries a produce indefinitely by default, so against a broker that stopped answering a background context has nothing to stop it: the flush interval elapsing, a cancelled run and a SIGTERM all wait for a broker that may never come back, and the supervisor kills the process instead. The pipeline's drain already hands the sink a context stripped of cancellation, so honouring the context here cannot cut the final write short.
func (*KafkaSink) Probe ¶ added in v1.1.0
Probe checks the broker before the first batch arrives.
Without it a wrong broker list produces a pipeline that starts normally, logs "consumer loop starting", and fails at the first flush. With a long flush interval that is minutes later, and a supervisor reports the pipeline healthy for every one of them. sinks.New probes any sink implementing Prober, so implementing it here is the whole change.
Ping asks the seed brokers for metadata, which is the cheapest request that proves one of them answered.
It runs off this goroutine because Ping does not reliably return when its context ends. Against a broker that accepts the connection and then never answers -- a partition rather than a refusal -- it was still running six seconds after a three second deadline. probe() runs before the pipeline has consumed anything, so a probe that cannot be bounded hangs the start instead of failing it, which is the failure the probe exists to prevent.
func (*KafkaSink) WriteTable ¶
WriteTable buffers the encoded rows. Nothing is produced here.
It used to call Produce for every row, so records reached the broker before any flush and a flush failure could not hold them back. The pipeline commits offsets on what Flush reports, so a sink that delivers earlier than it reports leaves the two out of step in the direction that loses rows.
type Option ¶ added in v1.0.5
type Option func(*options)
Option configures how a sink is built.
func WithMeterProvider ¶ added in v1.0.5
func WithMeterProvider(mp metric.MeterProvider) Option
WithMeterProvider supplies the provider the retry counter records through. Without one the counter records nothing, which is what a pipeline started with no --metrics wants.
func WithSinkRole ¶ added in v1.1.0
WithSinkRole names what this sink is for: "pipeline", "dlq" or "manager".
It separates the row-count series. Without it, DLQ rows sum into the same counter as delivered rows and the end-to-end ratio overstates delivery -- the metric would report the pipeline healthier the more records it rejected.
type Prober ¶ added in v1.0.5
Prober is implemented by a sink that can check its destination before the first batch arrives.
Without it a sink only discovers its destination when a batch reaches it. clickhouse.Open validates options and never dials, so a pipeline whose DSN names a host that does not resolve starts normally, logs "consumer loop starting", and runs. With a long flush interval the failure appears minutes later, and a supervisor calls the pipeline healthy for all of them.
type RetryPolicy ¶ added in v1.0.5
type RetryPolicy struct {
MaxAttempts int
InitialBackoff time.Duration
MaxBackoff time.Duration
Deadline time.Duration
}
RetryPolicy bounds how long a sink keeps trying a destination that is not answering.
Deadline bounds the whole ladder rather than one attempt, and it is the field that matters most. The retry runs inside the pipeline's open state transaction, and DuckDB's now() returns that transaction's start time, so a ladder that outlives the flush interval freezes the window clock -- the bug #158 fixed, reached by a second route.
func RetryPolicyFrom ¶ added in v1.0.5
func RetryPolicyFrom(c *config.SinkRetry) RetryPolicy
RetryPolicyFrom resolves a config block into a policy, filling in defaults for anything the user left out.
func (RetryPolicy) Enabled ¶ added in v1.0.5
func (p RetryPolicy) Enabled() bool
Enabled reports whether the policy will ever make a second attempt.
type SQLCommandSink ¶
type SQLCommandSink struct {
// contains filtered or unexported fields
}
SQLCommandSink materializes the result batch as a table and runs arbitrary DuckDB SQL against it. This is how the Python engine writes parquet, S3, Postgres, DuckLake and MotherDuck outputs.
func NewSQLCommandSink ¶
func NewSQLCommandSink(conn adbc.Connection, sql string, substitutions []config.SQLCommandSubstitution) (*SQLCommandSink, error)
func (*SQLCommandSink) BufferedRows ¶ added in v1.1.0
func (s *SQLCommandSink) BufferedRows() int
BufferedRows reports the rows this sink is holding that no flush has delivered. The pipeline publishes it as sink_buffered_rows, so an operator can tell a sink retrying a destination from one that has stopped draining.
func (*SQLCommandSink) Flush ¶
func (s *SQLCommandSink) Flush(ctx context.Context) error
Flush runs the sink SQL over the buffered rows, and keeps them buffered if it cannot.
Every batch used to be released on the way out whatever the outcome, so a failed flush left nothing to retry. Nothing calls Flush twice on this sink today -- retriesHelp excludes it, and processBatch calls it once and lets the error stop the pipeline -- so the rows replayed from the source rather than vanishing. That made it correct by accident of the call pattern, and the accident ends the day anything retries a sqlcommand sink, or an error policy makes a failed flush non-fatal.
Keeping the batch is the invariant. Whether a retry ladder is wrapped around this sink is a different question with a different answer, and retriesHelp answers only that one.
Source Files