# file: producer.aus

## class: KafkaProducer

[34:14] (extern: com.lehman.aussom.KafkaProducer) **extends: object** 

The Kafka producer. Configure with the typed setters and the
setConfig/setConfigs passthrough, then connect() builds the client:
prod = new KafkaProducer("broker1:9092");
prod.connect();
md = prod.send("orders.created", "o-1001", "{\"qty\": 3}");
send() is synchronous: it blocks on the broker acknowledgement
(bounded by delivery.timeout.ms) and returns the record metadata.
sendAsync() is fire-and-forget; failures are surfaced by flush()
and getLastAsyncError(). The producer is thread safe once
connected.

#### Methods

- **KafkaProducer** (`string Bootstrap`)

	> Creates a producer for a bootstrap address. Does not connect yet.

	- **@p** `Bootstrap` is a string with host:port[,host:port...].


- **newProducer** (`string Bootstrap`)


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


- **setAcks** (`string Acks`)

	> Sets the acknowledgement level.

	- **@p** `Acks` is a string: all (default), 1, or 0.
	- **@r** `this` object


- **setCompression** (`string Type`)

	> Sets the compression codec for record batches.

	- **@p** `Type` is a string: none, gzip, snappy, lz4, or zstd.
	- **@r** `this` object


- **setLingerMs** (`int Ms`)

	> Sets how long the producer waits to fill a batch before sending.

	- **@p** `Ms` is an int with the linger time in milliseconds.
	- **@r** `this` object


- **setBatchSize** (`int Bytes`)

	> Sets the maximum batch size.

	- **@p** `Bytes` is an int with the batch size in bytes.
	- **@r** `this` object


- **setEnableIdempotence** (`bool On`)

	> Enables or disables the idempotent producer (client default is enabled).

	- **@p** `On` is a bool.
	- **@r** `this` object


- **setTransactionalId** (`string Id`)

	> Sets the transactional id, enabling the transactional producer (initTransactions / beginTransaction / commitTransaction).

	- **@p** `Id` is a string with the transactional id.
	- **@r** `this` object


- **setDeliveryTimeoutMs** (`int Ms`)

	> Sets the upper bound on how long a synchronous send() blocks.

	- **@p** `Ms` is an int with the delivery timeout in milliseconds.
	- **@r** `this` object


- **setRequestTimeoutMs** (`int Ms`)

	> Sets the per-request broker timeout.

	- **@p** `Ms` is an int with the request timeout in milliseconds.
	- **@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 producer config key directly (security included). The ssl.keystore.location / ssl.truststore.location keys are rejected here - use setSslKeystoreFile / setSslTruststoreFile.

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


- **send** (`string Topic, Key, Value, map Headers = null, int Partition = -1, int TimestampMs = -1`)

	> Sends one record and blocks on the broker acknowledgement.

	- **@p** `Topic` is a string with the topic name.
	- **@p** `Key` is the record key (string/Buffer per format), or null.
	- **@p** `Value` is the record value, or null (a tombstone).
	- **@p** `Headers` is an optional map of header name to value.
	- **@p** `Partition` is an optional int to pin a partition (-1 lets the partitioner choose).
	- **@p** `TimestampMs` is an optional int record timestamp (-1 lets the broker/client stamp it).
	- **@r** `A` map { "topic", "partition", "offset", "timestamp" }.


- **sendAsync** (`string Topic, Key, Value, map Headers = null, int Partition = -1, int TimestampMs = -1`)

	> Sends one record fire-and-forget. Failures are counted and surfaced by flush() / getLastAsyncError().

	- **@p** `Topic` is a string with the topic name.
	- **@p** `Key` is the record key, or null.
	- **@p** `Value` is the record value, or null.
	- **@p** `Headers` is an optional map of header name to value.
	- **@p** `Partition` is an optional int to pin a partition.
	- **@p** `TimestampMs` is an optional int record timestamp.
	- **@r** `this` object


- **flush** ()

	> Flushes all buffered sends.

	- **@r** `An` int with the sendAsync failure count since the last flush (0 means everything was delivered).


- **getLastAsyncError** ()

	> Gets the text of the most recent sendAsync failure.

	- **@r** `A` string with the error text, or null.


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


- **metrics** ()

	> Gets the client metrics.

	- **@r** `A` map of metric group.name to numeric value.


- **initTransactions** ()

	> Initializes the transactional producer. Requires setTransactionalId before connect().

	- **@r** `this` object


- **beginTransaction** ()

	> Begins a transaction.

	- **@r** `this` object


- **commitTransaction** ()

	> Commits the open transaction.

	- **@r** `this` object


- **abortTransaction** ()

	> Aborts the open transaction.

	- **@r** `this` object


- **sendOffsetsToTransaction** (`list Offsets, object Consumer`)

	> Adds consumed offsets to the open transaction for exactly-once consume-transform-produce.

	- **@p** `Offsets` is a list of { "topic", "partition", "offset" } maps - each offset is the NEXT offset to read (last consumed + 1).
	- **@p** `Consumer` is the KafkaConsumer whose group consumed them.
	- **@r** `this` object


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

	> Closes the producer. 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.




