Class KafkaCoordinationTransport

java.lang.Object
io.krabka.streams.coordination.KafkaCoordinationTransport
All Implemented Interfaces:
CoordinationTransport, AutoCloseable
public final class KafkaCoordinationTransport extends Object implements CoordinationTransport
The CoordinationTransport that talks to a Kafka cluster.

The transport composes an Admin client, two kinds of producer, and one consumer. It adds no wire message. The points below are the parts of the composition that a reader cannot see from the call sites, and each one is load-bearing.

One role uses two producers

The transactional producer carries transactional.id = <role>, and it writes the lease records of that role. A second, plain producer appends the registration records of every role. Kafka puts every send of a transactional producer inside a transaction, and a candidate holds no epoch to open one with, so a registration cannot travel on the transactional producer.

The transactional producer is the epoch

acquireEpoch(Role) builds the transactional producer of the role and calls initTransactions. That call sends InitProducerId to the transaction coordinator, and the coordinator mints the producer epoch of the role. The transport keeps the producer, because the same instance stays bound to the epoch that the coordinator minted for it, and every write it makes carries that epoch. The Java producer reports no epoch of its own, so the transport reads the minted pair back with DescribeTransactions. A member that another member overtakes between the two calls reads the newer pair, and its first lease write then fails with FencedException. That is the same signal the design gives every loser.

A fence is the loss of the role

writeLease(Role, FencingToken, Lease) maps broker error code 47 INVALID_PRODUCER_EPOCH and broker error code 90 PRODUCER_FENCED onto FencedException. That mapping is the whole mechanism by which a deposed leader learns that it lost the role. No clock takes part in it. register(Role, MemberId, long) does not use the mapping. Its producer holds no epoch for the role.

All records of one role go to one partition

The succession rules rank candidates on the offset of their registration, so the records of one role need a total order, and one partition gives that order. Both producers pin the partition with RolePartitioner. See that class for why a key-hashing partitioner breaks the order.

The read is committed

readRoleRecords(Role) drives a caller-owned consumer with assign, seek, and poll, so the read needs no consumer group. Configure that consumer with isolation.level=read_committed, so an aborted lease write stays invisible. The read walks from the first offset of the partition to the end offset, and it drops every record whose key belongs to another role.

Example

try (Admin admin = Admin.create(adminSettings);
        Producer<byte[], byte[]> registrar = new KafkaProducer<>(plainSettings);
        Consumer<byte[], byte[]> reader = new KafkaConsumer<>(committedReadSettings);
        KafkaCoordinationTransport transport = new KafkaCoordinationTransport(
            admin, registrar, role -> new KafkaProducer<>(transactionalSettings(role)),
            reader)) {
    transport.register(role, me, System.currentTimeMillis());
}

  • Field Summary

    Fields
    Modifier and Type
    Field
    Description
    static final Duration
    DEFAULT_POLL_TIMEOUT
    The poll timeout a read uses when the caller names none.

  • Constructor Summary

    Constructors
    Constructor
    Description
    KafkaCoordinationTransport(org.apache.kafka.clients.admin.Admin admin, org.apache.kafka.clients.producer.Producer<byte[],byte[]> registrationProducer, Function<Role, org.apache.kafka.clients.producer.Producer<byte[],byte[]>> transactionalProducers, org.apache.kafka.clients.consumer.Consumer<byte[],byte[]> stateConsumer)
    Creates a transport with the default poll timeout.
    KafkaCoordinationTransport(org.apache.kafka.clients.admin.Admin admin, org.apache.kafka.clients.producer.Producer<byte[],byte[]> registrationProducer, Function<Role, org.apache.kafka.clients.producer.Producer<byte[],byte[]>> transactionalProducers, org.apache.kafka.clients.consumer.Consumer<byte[],byte[]> stateConsumer, Duration pollTimeout)
    Creates a transport.

  • Method Summary

    Modifier and Type
    Method
    Description
    FencingToken
    acquireEpoch(Role role)
    Mints a new epoch for a role and fences the member that held it.
    void
    clearLease(Role role, FencingToken token)
    Clears the lease of a role with a tombstone, in a transaction under one token.
    void
    close()
    Closes the transactional producers this transport built.
    Optional<FencingToken>
    describe(Role role)
    Asks the transaction coordinator which token holds a role now.
    List<CoordinationEntry>
    readRoleRecords(Role role)
    Reads the whole partition of a role and returns the records in offset order.
    void
    register(Role role, MemberId member, long registeredAtMillis)
    Appends the registration of one member to the partition of a role.
    void
    writeLease(Role role, FencingToken token, Lease lease)
    Writes the lease of a role in a transaction under one token.

    Methods inherited from class Object

    clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

  • Field Details

    • DEFAULT_POLL_TIMEOUT

      public static final Duration DEFAULT_POLL_TIMEOUT
      The poll timeout a read uses when the caller names none.

  • Constructor Details

    • KafkaCoordinationTransport

      public KafkaCoordinationTransport(org.apache.kafka.clients.admin.Admin admin, org.apache.kafka.clients.producer.Producer<byte[],byte[]> registrationProducer, Function<Role, org.apache.kafka.clients.producer.Producer<byte[],byte[]>> transactionalProducers, org.apache.kafka.clients.consumer.Consumer<byte[],byte[]> stateConsumer)
      Creates a transport with the default poll timeout.
      Parameters:
      admin - the client that answers DescribeTransactions
      registrationProducer - the plain producer that appends registration records
      transactionalProducers - builds the producer of one role, with transactional.id set to the role name
      stateConsumer - the consumer that reads CoordinationCodec.TOPIC
      Throws:
      NullPointerException - if an argument is null

    • KafkaCoordinationTransport

      public KafkaCoordinationTransport(org.apache.kafka.clients.admin.Admin admin, org.apache.kafka.clients.producer.Producer<byte[],byte[]> registrationProducer, Function<Role, org.apache.kafka.clients.producer.Producer<byte[],byte[]>> transactionalProducers, org.apache.kafka.clients.consumer.Consumer<byte[],byte[]> stateConsumer, Duration pollTimeout)
      Creates a transport.
      Parameters:
      admin - the client that answers DescribeTransactions
      registrationProducer - the plain producer that appends registration records
      transactionalProducers - builds the producer of one role, with transactional.id set to the role name
      stateConsumer - the consumer that reads CoordinationCodec.TOPIC
      pollTimeout - how long one poll of the coordination topic may block
      Throws:
      NullPointerException - if an argument is null

  • Method Details

    • acquireEpoch

      public FencingToken acquireEpoch(Role role)
      Mints a new epoch for a role and fences the member that held it.

      A Kafka producer calls initTransactions once, and the epoch it receives binds to that instance for its whole life. A second call for one role builds a second producer and closes the first. The token of the first call stops working at that point, which is correct: the coordinator superseded it.

      Specified by:
      acquireEpoch in interface CoordinationTransport
      Parameters:
      role - the role to take
      Returns:
      the token the transaction coordinator minted
      Throws:
      CoordinationException - if InitProducerId fails, or if the coordinator reports no transaction for the role after the call

    • readRoleRecords

      public List<CoordinationEntry> readRoleRecords(Role role)
      Reads the whole partition of a role and returns the records in offset order.
      Specified by:
      readRoleRecords in interface CoordinationTransport
      Parameters:
      role - the role to read
      Returns:
      the decoded records, oldest first
      Throws:
      CoordinationException - if the topic is absent, or if a record of this role does not decode

    • register

      public void register(Role role, MemberId member, long registeredAtMillis)
      Appends the registration of one member to the partition of a role.
      Specified by:
      register in interface CoordinationTransport
      Parameters:
      role - the role the member competes for
      member - the member that announces itself
      registeredAtMillis - the instant of the registration, in milliseconds since the Unix epoch
      Throws:
      CoordinationException - if the append fails

    • writeLease

      public void writeLease(Role role, FencingToken token, Lease lease)
      Writes the lease of a role in a transaction under one token.
      Specified by:
      writeLease in interface CoordinationTransport
      Parameters:
      role - the role the lease belongs to
      token - the token this transport minted for the role
      lease - the lease record to write
      Throws:
      FencedException - if the broker rejects the token
      CoordinationException - if this transport never minted the token, or if the write fails for another reason

    • clearLease

      public void clearLease(Role role, FencingToken token)
      Clears the lease of a role with a tombstone, in a transaction under one token.
      Specified by:
      clearLease in interface CoordinationTransport
      Parameters:
      role - the role to release
      token - the token this transport minted for the role
      Throws:
      FencedException - if the broker rejects the token
      CoordinationException - if this transport never minted the token, or if the write fails for another reason

    • describe

      public Optional<FencingToken> describe(Role role)
      Asks the transaction coordinator which token holds a role now.
      Specified by:
      describe in interface CoordinationTransport
      Parameters:
      role - the role to look up
      Returns:
      the current token, and empty when no member has ever taken the role
      Throws:
      CoordinationException - if the coordinator lookup fails

    • close

      public void close()
      Closes the transactional producers this transport built.

      The caller keeps the admin client, the registration producer, and the consumer, so this method leaves those three open.

      Specified by:
      close in interface AutoCloseable
      Specified by:
      close in interface CoordinationTransport