# file: shareconsumer.aus

## class: KafkaShareConsumer

[39:14] (extern: com.lehman.aussom.KafkaShareConsumer) **extends: object** 

The Kafka share consumer: queue semantics on a Kafka topic
(KIP-932, broker 4.2+). Many consumers in a share group can read
the same partitions; each record is individually accepted,
released (redelivered), or rejected (poison). The broker's
delivery-count limit archives records that fail too many times.
cons = new KafkaShareConsumer("broker1:9092", "billing");
cons.setAckMode("explicit");
cons.connect();
cons.subscribe([ "orders.created" ]);
recs = cons.poll(1000);
// ... handle, then per record:
cons.acknowledge(recs[0], "accept");
cons.commitSync();
Threading: same one-thread rule as KafkaConsumer; only wakeup()
is safe from another thread.

#### Methods

- **KafkaShareConsumer** (`string Bootstrap, string GroupId`)

	> Creates a share consumer. Does not connect yet.

	- **@p** `Bootstrap` is a string with host:port[,host:port...].
	- **@p** `GroupId` is a string with the share group id (required).


- **newShareConsumer** (`string Bootstrap, string GroupId`)


- **setAckMode** (`string Mode`)

	> Sets the acknowledgement mode.

	- **@p** `Mode` is a string: implicit (default; the next poll/commit accepts the previous batch) or explicit (every record must be acknowledged before the next poll).
	- **@r** `this` object


- **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


- **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 share consumer config key directly. 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.

	- **@r** `this` object


- **subscribe** (`list Topics`)

	> Subscribes to topics. Share groups have no pattern subscribe or manual assignment.

	- **@p** `Topics` is a list of topic name strings.
	- **@r** `this` object


- **unsubscribe** ()

	> Drops the subscription.

	- **@r** `this` object


- **subscription** ()

	> Gets the subscribed topics.

	- **@r** `A` list of topic name strings.


- **poll** (`int TimeoutMs`)

	> Polls for acquired records.

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


- **acknowledge** (`object Record, string Type = "accept"`)

	> Acknowledges one record from the last poll (explicit mode).

	- **@p** `Record` is the KafkaRecord to acknowledge.
	- **@p** `Type` is an optional string: accept (default), release (redeliver to someone), reject (poison, never redeliver), or renew (extend the acquisition lock).
	- **@r** `this` object


- **commitSync** (`int TimeoutMs = -1`)

	> Commits the acknowledgements to the broker synchronously.

	- **@p** `TimeoutMs` is an optional int bound on the commit wait.
	- **@r** `A` list of { "topic", "partition", "error" } maps - empty when every acknowledgement was applied.


- **commitAsync** ()

	> Commits the acknowledgements asynchronously.

	- **@r** `this` object


- **setOnAcknowledgementCommit** (`callback Cb`)

	> Sets a callback invoked on this thread (from poll/commit/ close) as acknowledgement commits complete.

	- **@p** `Cb` is a callback taking (list Entries, string ErrText); each entry is { "topic", "partition", "offsets" } and ErrText is null on success.
	- **@r** `this` object


- **acquisitionLockTimeoutMs** ()

	> Gets the broker's record acquisition lock timeout once the client has learned it.

	- **@r** `An` int with the timeout in milliseconds, or null.


- **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 call.

	- **@r** `this` object


- **close** (`int TimeoutMs = -1`)

	> Closes the share consumer, releasing unacknowledged records. 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.




