kafka 1.0.0

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.