# file: consumer.aus

## 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.




