kafka 1.0.0

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.