KAFKA-15213: provide the exact offset to QuorumController.replay

[cmccabe]

Provide the exact record offset to QuorumController.replay() in all cases. There are several situations
where this is useful, such as logging, implementing metadata transactions, or handling broker
registration records.

In the case where the QC is inactive, and simply replaying records, it is easy to compute the exact
record offset from the batch base offset and the record index.

The active QC case is more difficult. Technically, when we submit records to the Raft layer, it can
choose a batch end offset later than the one we expect, if someone else is also adding records.
While the QC is the only entity submitting data records, control records may be added at any time.
In the current implementation, these are really only used for leadership elections. However, this
could change with the addition of quorum reconfiguration or similar features.

Therefore, this PR allows the QC to tell the Raft layer that a record append should fail if it
would have resulted in a batch end offset other than what was expected. This in turn will trigger a
controller failover. In the future, if automatically added control records become more common, we
may wish to have a more sophisticated system than this simple optimistic concurrency mechanism. But
for now, this will allow us to rely on the offset as correct.

In order that the active QC can learn what offset to start writing at, the PR also adds a new
endOffset parameter to handleLeaderChange. Since the Raft layer only invokes handleLeaderChange on
the active once it has replayed the log, this information should always be up-to-date in that
context.

At the Raft level, this PR adds a new exception, UnexpectedEndOffsetException. This gets thrown
when we request an end offset that doesn't match the one the Raft layer would have given us.
Although this exception should cause a failover, it should not be considered a fault. This
complicated the exception handling a bit and motivated splitting more of it out into the new
EventHandlerExceptionInfo class. This will also let us unit test things like slf4j log messages a
bit better.

[mumrah]

Thanks @cmccabe, left some questions and minor nits. I think the overall approach is fine.

Regarding the failure case where the end offset of a batch is not equal to what the controller expected, will this only happen if a raft election occurred that was not initiated by a resignation? What are the circumstances when this can happen? Raft timeouts?

IIUC, when an election happens in the middle of an atomic batch, the batch will be lost anyways. The node will finish writing the batch to the local log at epoch N, then process the new leader at N+1, and then it will truncate its own log once it fetches from the new leader at N+1 and sees the start offset for the epoch is less than its own end offset. Is that about right?

[mumrah]

After an election, is the endOffset report by raft both the last written and last committed offset?

[cmccabe]

The endOffset comes directly from the log and is as described... the end offset (exclusive).

Thinking about it more, I don't think I need to assume that it's committed, so I won't.

But it is used to calculate the next offset that the active controller should try to write to.

[cmccabe]

Regarding the failure case where the end offset of a batch is not equal to what the controller expected, will this only happen if a raft election occurred that was not initiated by a resignation? What are the circumstances when this can happen? Raft timeouts?

IIUC, when an election happens in the middle of an atomic batch, the batch will be lost anyways. The node will finish writing the batch to the local log at epoch N, then process the new leader at N+1, and then it will truncate its own log once it fetches from the new leader at N+1 and sees the start offset for the epoch is less than its own end offset. Is that about right?

There's been some discussion of adding more Raft internal control records. One example is if we wanted to implement a dynamic change-of-quorum mechanism. There would probably be internal Raft records associated with that. It's not clear whether change-of-quorum events would also always involve a leader change -- I think in some cases they would not.

Like I said earlier, if we end up adding more background raft messages, we might introduce some mechanism for the active controller to get an "offset lock" so it can get an offset, replay the records, and then try to commit them under that lock. That would minimize failovers caused by these background messages. But since they don't exist today, we can avoid that for today.

[mumrah]

Thanks for the explanation. As you mentioned earlier we are basically doing optimistic concurrency control by doing the atomic append, then checking the resulting state (i.e., offset). With transactions we'll have the ability to roll back metadata state to a particular offset with the abort marker record.

If a non-transactional write from the controller happens and the expected end offset check fails, the controller will resign. I believe that batch will still be committed and processed by the quorum, right?. I think this is correct, but I just want to make sure this scenario is well understood.

[jsancio]

@cmccabe This doesn't look like a MINOR change since you are changing the interface of a few raft types. Do you mind creating a Jira that explains the problem you are trying to solve?

[jsancio]

This is only possible in the leader. This is not well defined on any replica outside of the leader. What problem are you trying so solve?

KRaft guarantees that the committed offset is the LEO when notifying the leader that is now the leader of the partition. For all other replicas leadership change and lost of leadership is done immediately.

[cmccabe]

What problem are you trying so solve?

We have to know the offset of records that we apply. But we apply records before we submit them to Raft.

[jsancio]

This is the same base or starting offset. The RaftClient does expose LEO but it does expose a base offset. I think we should do something like this:

     * @param expectedBaseOffset if this is set, it matches the base offset that KRaft will use for the {@code records}.

[cmccabe]

ok. we can use requiredBaseOffset.

[jsancio]

Please document this method.

[cmccabe]

ok

[jsancio]

If the user of RaftClient provides the "expected base offset" this becomes expectedBaseOffset.ifPresent(baseOffset -> if (baseOffset == nextOffset) { ... } );

[jsancio]

@cmccabe Why do you need this change now? Since KRaft is a single writer (The active controller) and the active controller is guarantee to have see all of the records before becoming leader, don't you always know what is going to be the next offset?

[cmccabe]

@cmccabe This doesn't look like a MINOR change since you are changing the interface of a few raft types. Do you mind creating a Jira that explains the problem you are trying to solve?

OK. @jsancio , I filed https://issues.apache.org/jira/browse/KAFKA-15213

[cmccabe]

@cmccabe Why do you need this change now? Since KRaft is a single writer (The active controller) and the active controller is guarantee to have see all of the records before becoming leader, don't you always know what is going to be the next offset?

It is possible for Raft itself to insert control records. So it is not enough to rely on guessing the next offset. We have to know for sure.

[jsancio]

@cmccabe Why do you need this change now? Since KRaft is a single writer (The active controller) and the active controller is guarantee to have see all of the records before becoming leader, don't you always know what is going to be the next offset?

It is possible for Raft itself to insert control records. So it is not enough to rely on guessing the next offset. We have to know for sure.

It is not possible in the current implementation. The active controller will lose leadership if KRaft inserts a control record. Have you seen this in practice?

[cmccabe]

It is not possible in the current implementation. The active controller will lose leadership if KRaft inserts a control record. Have you seen this in practice?

The implementation could change. If we're going to rely on the offsets for correctness, we need guarantees.

[jsancio]

@cmccabe @mumrah and I discuss this PR offline. We agreed to add long RaftClient::logEndOffset() method that would return

1. The BatchAccumulator's baseOffset when the replica is a leader or
2. The log end offset when the replica is not a leader.

This is instead of the sending the LEO in the RaftClient.Listener::handleLeaderChange callback.

[cmccabe]

Thanks, @jsancio . I updated the PR with the results of our discussion. One small change I made is that I think RaftClient#logEndOffset can just unconditionally return the log end offset, as the name implies. We only need to call it when becoming active anyway.

[jsancio]

Thanks for the changes @cmccabe

[jsancio]

This indentation doesn't look right. We indent 4 spaces in this case.

[jsancio]

This is not correct in all cases. The leader can have records in the base accumulator that have not been sent to the log. I think you want something along these lines:

    public long logEndOffset() {
        return quorum.maybeLeaderState()
            .map(leader -> leader.accumulator().nextOffset())
            .orElse(log.endOffset().offset);
    }

Then we can add this method to BatchAccumulator:

    public long nextOffset() {
        appendLock.lock();
        try {
            return nextOffset;
        finally {
            appendLock.unlock();
        }
    }

[jsancio]

Can we also add tests for this new functionality?

[cmccabe]

This is not correct in all cases. The leader can have records in the base accumulator that have not been sent to the log.

Hmm... the method name is "logEndOffset." So it should just return the log end offset, right? Returning something else would be misleading.

In any case, we only need this method when the leader becomes active. We will not use it after that.

[cmccabe]

Can we also add tests for this new functionality?

Yes, good point. I will add a test for the requiredEndOffset parameter and the logEndOffset method.

[jsancio]

Hmm... the method name is "logEndOffset." So it should just return the log end offset, right? Returning something else would be misleading.

@cmccabe, I don't follow this comment. When the client calls KafkaRaftClient::schedule{Atomic}Append the KafkaRaftClient compare the provided offset with the nextOffset stored in the BatchAccumulator. If we want this method to succeed in most cases KafkaRaftClient::logEndOffset should return that offset, BatchAccumulator::nextOffset and not the log end offset.

Maybe logEndOffset is not a great name. I am okay renaming this to KafkaRaftClient::endOffset() but I am open to suggestions.

[jsancio]

I think must readers will assume that this "end offset" is exclusive. I think this offset is inclusive. We normally use lastOffset for this kind of offset.

            long lastOffset = nextOffset + records.size() - 1;

[jsancio]

We need tests for the new functionality added to the BatchAccumulator.

[jsancio]

We need tests for the new functionality added to KafkaRaftClient. That is both the new method logEndOffset and the changes to scheduleAtomicAppend.

[jsancio]

I couldn't find a test for the new KafkaRaftClient::scheduleAtomicAppend.

[mumrah]

It's probably worth adding a sentence or two about the new optimistic concurrency.

[mumrah]

Previously, we would update lastCommittedOffset as we got the handleCommit callback from our RaftClient. Since we process Raft events sequentially (and they are delivered sequentially from a single thread), we always process any commit callbacks before a leader change. Which means this offset is valid with respect to the end offset when the leadership changed.

Now, while we're processing a leader change, we ask RaftClient for its end offset. Is there any possibility that commits could be made that would make this end offset greater than we expect? Basically, can we be sure that the end offset doesn't change between the time Raft becomes the leader and this event is processed?

[cmccabe]

Well, there is not really any difference between the previous iterations of this PR and the current one in this regard. If some other component that isn't the controller is adding messages, our supplied requiredBaseOffset may be invalid. It is only a snapshot of the offset at a point in time, after all. Which is why we check requiredBaseOffset in atomicAppend.

[mumrah]

Understood, I just wanted to make sure I understood the behavior here regarding the new RaftClient API. It sounds like it's fine for us as long as we're single writer.

[mumrah]

LGTM

[jsancio]

Thanks for the changes @cmccabe .

[jsancio]

Hmm... the method name is "logEndOffset." So it should just return the log end offset, right? Returning something else would be misleading.

@cmccabe, I don't follow this comment. When the client calls KafkaRaftClient::schedule{Atomic}Append the KafkaRaftClient compare the provided offset with the nextOffset stored in the BatchAccumulator. If we want this method to succeed in most cases KafkaRaftClient::logEndOffset should return that offset, BatchAccumulator::nextOffset and not the log end offset.

Maybe logEndOffset is not a great name. I am okay renaming this to KafkaRaftClient::endOffset() but I am open to suggestions.

[jsancio]

What's the argument/reason for adding this functionality to scheduleAtomicAppend and not scheduleAppend?

[cmccabe]

The controller doesn't use scheduleAppend. We only use scheduleAtomicAppend.

[jsancio]

I couldn't find a test for the new KafkaRaftClient::scheduleAtomicAppend.

[cmccabe]

@cmccabe, I don't follow this comment. When the client calls KafkaRaftClient::schedule{Atomic}Append the KafkaRaftClient compare the provided offset with the nextOffset stored in the BatchAccumulator. If we want this method to succeed in most cases KafkaRaftClient::logEndOffset should return that offset, BatchAccumulator::nextOffset and not the log end offset.

There can't be anything in the accumulator when we become active, because we are not adding things to the accumulator when we are inactive. Therefore all we need to know is the end of the log.

After transitioning from inactive to active, the controller tracks its own offset and will not invoke this method. So assuming leadership never moves, the method would be invoked only once ever.

[cmccabe]

I couldn't find a test for the new KafkaRaftClient::scheduleAtomicAppend.

I added KafkaRaftClientTest.testAppendWithRequiredBaseOffset.