Class CoordinationCodec

java.lang.Object
io.krabka.streams.coordination.CoordinationCodec
public final class CoordinationCodec extends Object
Encodes and decodes the frozen records of the internal __coordination_state topic.

A member of a coordination role writes its registration and its lease to this compacted topic. The key carries the record kind, so one topic holds both kinds. The key is also the compaction key, so the topic keeps the last record of every role and every member. A record with a null value is a tombstone. A tombstone of a registration key deregisters one member, and a tombstone of a lease key clears the lease of one role.

The layouts below are frozen. krabka-client-rs and krabka-streams-go implement them byte for byte, and all three projects assert the same golden bytes. Do not change a field, a field order, or a version number without the same change in the two ports. 

key:
  version  i16 = 0
  kind     i16          0 registration, 1 lease
  role     string
  member   string       the member id for kind 0, the empty string for kind 1

registration value:
  version        i16 = 0
  member         string
  registered_at  i64    epoch milliseconds

lease value:
  version         i16 = 0
  member          string
  producer_id     i64
  producer_epoch  i16
  granted_at      i64   epoch milliseconds
  deadline        i64   epoch milliseconds

Every integer is big-endian and signed. A string is an i16 byte length and then plain UTF-8 bytes. This is Kafka's own native string layout, and it is the layout that BarrierCutDecoder documents for __barrier_state. It is not Java's modified UTF-8, which DataOutput.writeUTF produces. The length is never negative, because the format has no null string. An absent member is the empty string, and it encodes as the two bytes 00 00.

Malformed bytes raise CoordinationException. The decoder never returns a partly filled record, and it never sizes an allocation from a length it read.

Example

byte[] key = CoordinationCodec.encodeKey(CoordinationKey.lease(role));
byte[] value = CoordinationCodec.encodeValue(lease);
producer.send(new ProducerRecord<>(CoordinationCodec.TOPIC, partition, key, value));

CoordinationKey decoded = CoordinationCodec.decodeKey(key);
Optional<CoordinationValue> back = CoordinationCodec.decodeValue(decoded.kind(), value);

  • Field Details

    • TOPIC

      public static final String TOPIC
      The compacted internal topic that holds the coordination state of a cluster.

      Every record of one role goes to one partition, so the records of a role carry a total order. See RolePartitioner. The topic needs cleanup.policy=compact. A client writes it with acks=all, and an operator should set min.insync.replicas to at least 2.

      See Also:

  • Method Details

    • encodeKey

      public static byte[] encodeKey(CoordinationKey key)
      Encodes a record key.

      The method writes the empty member string for a lease key, because the frozen layout gives a lease no member.

      Parameters:
      key - the key to encode
      Returns:
      the key bytes of the frozen layout

    • encodeValue

      public static byte[] encodeValue(CoordinationValue value)
      Encodes a record value.

      A tombstone has no encoded form. Send a null value to write one.

      Parameters:
      value - the registration or the lease to encode
      Returns:
      the value bytes of the frozen layout

    • decodeKey

      public static CoordinationKey decodeKey(byte[] bytes)
      Decodes a record key.
      Parameters:
      bytes - the key bytes as they arrived from the broker
      Returns:
      the decoded key
      Throws:
      CoordinationException - if the buffer is truncated, if it holds trailing bytes, if the version is not 0, if the kind is neither 0 nor 1, if a string length is negative, if a string is not UTF-8, if a lease key names a member, or if the role or the member breaks its bounds
      NullPointerException - if the key bytes are null

    • decodeValue

      public static Optional<CoordinationValue> decodeValue(RecordKind kind, byte[] bytes)
      Decodes a record value under the kind of its key.
      Parameters:
      kind - the kind the key carried
      bytes - the value bytes, or null for a tombstone
      Returns:
      the decoded value, and empty for a tombstone
      Throws:
      CoordinationException - if the buffer is truncated, if it holds trailing bytes, if the version is not 0, if a string length is negative, if a string is not UTF-8, if the member breaks its bounds, or if a lease value carries a negative producer id or producer epoch
      NullPointerException - if the kind is null

    • isTombstone

      public static boolean isTombstone(byte[] value)
      Reports whether a record value is a tombstone.

      No value of this topic is legally empty, so a null value and a zero-length value are both tombstones.

      Parameters:
      value - the record value bytes, or null
      Returns:
      true when the record clears its key