Walk through the SourceTask.poll() loop and the SinkTask.put()/flush() loop. What is each responsible for, and how do offsets get committed?

## Source side: the poll() loop A `SourceTask` runs on its own thread inside a worker. The framework loops: ``` while (running) { List<SourceRecord> records = task.poll(); // your code for (record : records) producer.send(record); // after acks, commit source offsets } ``` ### What poll() must do - Return a `List<SourceRecord>` of *new* data fetched from the external system since last time. - **Block** when there's nothing new (e.g. wait on a queue / sleep briefly), returning null or an empty list rather than spinning hot. - Embed two maps in each record: **sourcePartition** (identifies a logical stream, e.g. `{"filename":"a.log"}`) and **sourceOffset** (position within it, e.g. `{"position":1024}`). ### Source offset commit Connect does NOT use Kafka consumer offsets for sources (there's no consumer). Instead it produces the records to the destination topic, and once the producer acknowledges them, it writes the (sourcePartition -> sourceOffset) pairs to the internal **connect-offsets** topic (`offset.storage.topic`). On restart, `SourceTaskContext.offsetStorageReader()` lets the task read back the last committed offset and resume. `offset.flush.interval.ms` controls how often offsets are flushed. `commitRecord()` is an optional callback after each record is acked. ## Sink side: the put()/flush() loop A `SinkTask` is backed by a Kafka **consumer** managed by Connect. The framework loops: ``` while (running) { ConsumerRecords r = consumer.poll(); task.put(toSinkRecords(r)); // your code: write/buffer downstream if (timeToCommit) { offsets = task.preCommit(currentOffsets); // your code: flush + return safe offsets consumer.commitSync(offsets); } } ``` ### put(Collection<SinkRecord>) - Receives a batch of `SinkRecord`s. The task writes them to the external system, or buffers them for batched writes. - Must handle retries; throwing `RetriableException` makes Connect retry the same batch. ### flush() / preCommit() - On a periodic boundary (`offset.flush.interval.ms`), Connect calls `flush(currentOffsets)` or, preferably, `preCommit(currentOffsets)`. - The task must **make all buffered records durable downstream** before returning. - `preCommit()` returns the map of `TopicPartition -> OffsetAndMetadata` that the task has *actually* persisted. Connect commits exactly those to Kafka. Returning a lower offset than `currentOffsets` is how a task says "I've only safely written up to here." Returning an empty map disables Connect-managed offset commit (used when the sink stores offsets itself, e.g. for exactly-once into the external store). ## Why the order matters (delivery semantics) The golden rule: **persist downstream, THEN commit the Kafka offset.** If the task committed first and then crashed before writing, data would be lost (less than at-least-once). By writing first and committing the safe offset in preCommit(), Connect guarantees **at-least-once**. To reach **effectively/exactly-once**, the sink must be idempotent or transactional (e.g. keyed upserts, or storing offsets atomically with the data). For sources, exactly-once is available via the dedicated exactly-once source support (KIP-618: `exactly.once.source.support`) which wraps producer transactions around record batches and offset writes. ## Common pitfalls - Busy-spinning in poll() instead of blocking -> CPU burn. - Committing Kafka offsets in put() rather than after durable write -> data loss window. - Ignoring RetriableException vs a fatal exception -> unnecessary task FAILED state. - Not implementing preCommit() so offsets advance past not-yet-flushed data.