kafka 1.0.0

class: KafkaConsumer

[41:14] (extern: com.lehman.aussom.KafkaConsumer) extends: object

The Kafka consumer, for direct synchronous polling: cons = new KafkaConsumer("broker1:9092", "billing"); cons.setAutoOffsetReset("earliest"); cons.connect(); cons.subscribe([ "orders.created" ]); recs = cons.poll(1000); For continuous consumption under Aussom Server use KafkaListener instead - the server then owns the thread and lifecycle. Threading: a Kafka consumer belongs to ONE thread. The wrapper adds no locking; the Kafka client rejects cross-thread use itself. The single exception is wakeup(), callable from any thread to break a blocked poll(). Topic-partition maps are { "topic": s, "partition": i }; offsets lists are lists of those maps with an added "offset" (and optional "metadata") - the same shapes everywhere in this module.

Methods

• KafkaConsumer (string Bootstrap, string GroupId = null)

  Creates a consumer for a bootstrap address. Does not connect yet.

  • @p Bootstrap is a string with host:port[,host:port...].
  • @p GroupId is an optional string with the consumer group id (not needed for assign()-only use).

• newConsumer (string Bootstrap, string GroupId = null)

• setClientId (string Id)

  Sets the client id used in broker logs, metrics, and quotas.

  • @p Id is a string with the client id.
  • @r this object

• setAutoOffsetReset (string Mode)

  Sets where to start when the group has no committed offset.

  • @p Mode is a string: earliest, latest (default), or none.
  • @r this object

• setEnableAutoCommit (bool On)

  Enables or disables client auto-commit during poll.

  • @p On is a bool.
  • @r this object

• setMaxPollRecords (int N)

  Sets the maximum records one poll() returns.

  • @p N is an int with the record cap.
  • @r this object

• setMaxPollIntervalMs (int Ms)

  Sets the processing deadline between polls before the group evicts this consumer.

  • @p Ms is an int with the interval in milliseconds.
  • @r this object

• setSessionTimeoutMs (int Ms)

  Sets the group session timeout.

  • @p Ms is an int with the timeout in milliseconds.
  • @r this object

• setIsolationLevel (string Level)

  Sets the transactional read isolation.

  • @p Level is a string: read_uncommitted (default) or read_committed.
  • @r this object

• setKeyFormat (string Fmt)

  Sets the key format.

  • @p Fmt is a string: string (default) or bytes (Buffer).
  • @r this object

• setValueFormat (string Fmt)

  Sets the value format.

  • @p Fmt is a string: string (default) or bytes (Buffer).
  • @r this object

• setConfig (string Key, Value)

  Sets any consumer config key directly (group.protocol, group.instance.id, security, and the rest). The ssl.*.location keys are rejected here - use the dedicated setters.

  • @p Key is a string with the config key.
  • @p Value is the config value.
  • @r this object

• setConfigs (map Cfg)

  Applies a whole config map at once.

  • @p Cfg is a map of config key to value.
  • @r this object

• setSslKeystoreFile (string Path)

  Sets the TLS keystore file. Under Aussom Server the path is resolved through the app_data sandbox.

  • @p Path is a string with the app-relative keystore path.
  • @r this object

• setSslTruststoreFile (string Path)

  Sets the TLS truststore file. Under Aussom Server the path is resolved through the app_data sandbox.

  • @p Path is a string with the app-relative truststore path.
  • @r this object

• connect ()

  Builds the Kafka client from the stored config. Setters error after this point.

  • @r this object

• subscribe (list Topics, callback OnAssigned = null, callback OnRevoked = null)

  Subscribes to topics. The optional callbacks fire on this thread from inside poll() when the group rebalances, each with one list of topic-partition maps.

  • @p Topics is a list of topic name strings.
  • @p OnAssigned is an optional callback(list Partitions).
  • @p OnRevoked is an optional callback(list Partitions).
  • @r this object

• subscribePattern (string Regex, callback OnAssigned = null, callback OnRevoked = null)

  Subscribes to every topic matching a regex pattern.

  • @p Regex is a string with the topic pattern.
  • @p OnAssigned is an optional callback(list Partitions).
  • @p OnRevoked is an optional callback(list Partitions).
  • @r this object

• assign (list TopicPartitions)

  Manually assigns partitions - no group coordination, no rebalancing.

  • @p TopicPartitions is a list of topic-partition maps.
  • @r this object

• unsubscribe ()

  Drops the current subscription or assignment.

  • @r this object

• poll (int TimeoutMs)

  Polls for records.

  • @p TimeoutMs is an int with the maximum wait.
  • @r A list of KafkaRecord objects (possibly empty).

• commitSync (list Offsets = null, int TimeoutMs = -1)

  Commits offsets synchronously. With no arguments commits the last poll.

  • @p Offsets is an optional offsets list to commit exactly.
  • @p TimeoutMs is an optional int bound on the commit wait.
  • @r this object

• commitAsync (callback OnComplete = null)

  Commits the last poll asynchronously. The optional callback is invoked from within a later poll() on this thread.

  • @p OnComplete is an optional callback(list Offsets, string ErrText) - ErrText is null on success.
  • @r this object

• seek (string Topic, int Partition, int Offset)

  Seeks a partition to an offset for the next poll.

  • @p Topic is a string with the topic name.
  • @p Partition is an int with the partition.
  • @p Offset is an int with the offset.
  • @r this object

• seekToBeginning (list TopicPartitions)

  Seeks partitions back to their first record.

  • @p TopicPartitions is a list of topic-partition maps.
  • @r this object

• seekToEnd (list TopicPartitions)

  Seeks partitions to their end.

  • @p TopicPartitions is a list of topic-partition maps.
  • @r this object

• position (map TopicPartition)

  Gets the next-fetch position for one partition.

  • @p TopicPartition is a topic-partition map.
  • @r An int with the offset of the next record to fetch.

• committed (list TopicPartitions)

  Gets the committed offsets for partitions. Partitions with no committed offset are omitted from the result.

  • @p TopicPartitions is a list of topic-partition maps.
  • @r An offsets list.

• currentLag (map TopicPartition)

  Gets this consumer's lag on one partition.

  • @p TopicPartition is a topic-partition map.
  • @r An int with the lag, or null when not yet known.

• beginningOffsets (list TopicPartitions)

  Gets the earliest available offsets.

  • @p TopicPartitions is a list of topic-partition maps.
  • @r An offsets list.

• endOffsets (list TopicPartitions)

  Gets the latest offsets (the offset the next produced record would get).

  • @p TopicPartitions is a list of topic-partition maps.
  • @r An offsets list.

• offsetsForTimes (list Query)

  Finds the earliest offset at or after a timestamp, per partition. Partitions with no such record are omitted.

  • @p Query is a list of { "topic", "partition", "timestamp" } maps.
  • @r An offsets list with a "timestamp" on each entry.

• pause (list TopicPartitions)

  Pauses fetching from partitions. poll() keeps heartbeating so the consumer stays in its group.

  • @p TopicPartitions is a list of topic-partition maps.
  • @r this object

• resume (list TopicPartitions)

  Resumes fetching from paused partitions.

  • @p TopicPartitions is a list of topic-partition maps.
  • @r this object

• paused ()

  Gets the currently paused partitions.

  • @r A list of topic-partition maps.

• assignment ()

  Gets the currently assigned partitions.

  • @r A list of topic-partition maps.

• subscription ()

  Gets the subscribed topics.

  • @r A list of topic name strings.

• listTopics ()

  Lists all topics visible to this consumer.

  • @r A map of topic name to partition info list.

• partitionsFor (string Topic)

  Gets partition info for a topic.

  • @p Topic is a string with the topic name.
  • @r A list of { "topic", "partition", "leader", "replicas", "isr" } maps.

• enforceRebalance ()

  Forces a group rebalance on the next poll.

  • @r this object

• groupMetadata ()

  Gets the consumer group metadata.

  • @r A map { "groupId", "generationId", "memberId", "groupInstanceId" }.

• metrics ()

  Gets the client metrics.

  • @r A map of metric group.name to numeric value.

• wakeup ()

  Breaks a blocked poll() from another thread - the one thread-safe consumer call.

  • @r this object

• close (int TimeoutMs = -1)

  Closes the consumer, leaving the group cleanly. Idempotent.

  • @p TimeoutMs is an optional int bound on the close wait.
  • @r this object

• isConnected ()

  Reports whether connect() has been called.

  • @r A bool with true when connected.