...
lease_pid_request => topic partition pidResponse:
lease_pid_response => error pid generation sequence_number expire_ms
These requests This request could also have a batch version for handling multiple partitions at once, but I described the single partition case for simplicity.
There are several intended uses for this API:
...
Instead this proposal assumes the cluster will have some configured period of disuse after fixed lifetime for pids from the point of issuance after which a pid is available for reuse. It would also be possible to allow clients to define custom expirations in their lease_pid request but that would require a more complex implementation as all replications would have to know about each expiration. The server will issue pids approximately in order so reuse will only actually occur after 4 billion pids have been issued.
...
Note that the expiration is only approximate as it is based on the time a server sees the first message for a partition. However it is only required that the server guarantee at least that much time, so retaining pids longer is okay. This means the followers can use arrival time (though arrival on followers will be slightly older than on the leader). In the event of a full data restore the circular buffer of pid entries will be full and all will have full expiration time restored.
Client Implications
The general deduplication will happen automatically in the producer. It should be cheap and easy enough to enable by default.
To integrate this in tools like mirror maker and samza that chain producers and consumers we will need to be able to save the PID and sequence number of a producer. We can do this by including this in the response returned by the producer.
The producer will need a config to set it's initial PID, sequence number, and generation at initialization time.
We may want to consider extending the OffsetCommit request to also store these fields.