DVKafkaQueueAdapter class

DV.Queues on Kafka.

Kafka is a log, not a queue, and the difference decides everything here. A record is not removed when it is read: consumption is a committed offset, so "acknowledged" means the offset moved past it and a job cannot be individually deleted, retried out of order, or left for another worker.

That makes some of the queue contract impossible rather than merely unimplemented, and this adapter says so instead of approximating:

  • a failure cannot hand one record back while others move on, so a failed job is republished to the end of the topic and the offset advances -- the retry is a new record, which is the only thing the log allows;
  • there is no dead-letter list, because a dead letter would be a record the offset skipped and nothing records that it was skipped;
  • priority does not exist in a log whose order is arrival order.

One partition per queue. Ordering across partitions is not defined, and a queue that silently reordered work would be worse than a slow one.

Implemented types

Constructors

DVKafkaQueueAdapter(DVKafkaClient client, {String topicPrefix = 'dartvel.'})

Properties

client → DVKafkaClient
final
hashCode → int
The hash code for this object.
no setterinherited
runtimeType → Type
A representation of the runtime type of the object.
no setterinherited
topicPrefix → String
Prefix isolating this application's topics on a shared cluster.
final

Methods

complete(String id) → Future<void>
override
deadLetters(String queue) → Future<List<DVJobEnvelope<DVJobPayload>>>
override
discard(String id) → Future<bool>
Drops one dead-lettered job. Flushing a queue to be rid of a single poison message would take every other job with it.
override
enqueue<TPayload>(String queue, TPayload payload, {int priority = 0, int maxAttempts = 3, Duration backoff = const Duration(seconds: 30), String? tenant}) → Future<DVJobEnvelope<TPayload>>
Stores payload on queue.
override
fail(String id, String error, StackTrace stackTrace) → Future<void>
override
flush(String queue) → Future<int>
override
lag(String queue) → Future<int>
How far behind the end of the log this queue is.
noSuchMethod(Invocation invocation) → dynamic
Invoked when a nonexistent method or property is accessed.
inherited
offsetFor(String jobId) → int?
The offset that will be committed for jobId.
pending(String queue) → Future<List<DVJobEnvelope<DVJobPayload>>>
override
reserve(String queue) → Future<DVJobEnvelope<DVJobPayload>?>
override
retry(String id) → Future<bool>
override
topicFor(String queue) → String
toString() → String
A string representation of this object.
inherited

Operators

operator ==(Object other) → bool
The equality operator.
inherited