KAFKA-10394: generate snapshot

[jsancio]

This PR adds support for generating snapshot for KIP-630.

1. Adds the interfaces RawSnapshotWriter and RawSnapshotReader and the implementations FileRawSnapshotWriter and FileRawSnapshotReader respectively. These interfaces and implementations are low level API for writing and reading snapshots. They are internal to the Raft implementation and are not exposed to the users of RaftClient. They operation at the Record level. These types are exposed to the RaftClient through the ReplicatedLog interface.

2. Adds a buffered snapshot writer: SnapshotWriter<T>. This type is a higher-level type and it is exposed through the RaftClient interface. A future PR will add the related SnapshotReader<T>, which will be used by the state machine to load a snapshot.

More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.

Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[hachikuji]

Thanks for the PR. Left a few initial questions.

[hachikuji]

It would be nice if we can figure out how to consolidate this and BatchReader. It seems like it should be doable.

[jsancio]

This comment applies to some of your other observations. At high-level there are 4 use cases that we need to design and implement. Two use cases are for the Raft implementation. Two use cases are for the state machine.

Raft implementation

These types are internal to the raft implementation and don't have to be exposed to the state machine.

Leader Use Case

The leader needs to be able to send a part of the snapshot over the network. Something like this

interface SnapshotReader extends Closeable {
    long transferTo(long position, long maxBytes, WritableChannel channel);
    int read(ByteBuffer buffer, long position);
}

Follower Use Case

The followers need to be able to copy bytes from the network and validate the snapshot on disk when fetching is done.

interface SnapshotWriter extends Closeable {
    void append(ByteBuffer buffer);
    void validate();
    void freeze();
}

State machine implementation

These types are exposed to the state machine.

Load Snapshot

The state machine needs to be able to load/scan the entire snapshot. The state machine can use close to tell the raft client that it finished loading the snapshot. This will be implemented in a future PR but it could look like this:

interface BatchedSnapshotReader<T> extends Iterable<Iterable<T>>, Closeable {
}

Generate Snapshot

The state machine needs to be able to generate a snapshot by appending records/values and marking the snapshot as immutable (freeze) when it is done.

interface BatchdSnapshotWriter<T> extends Closeable {
   void append(Iterable<T> records);
   void freeze();
}

Notes

SnapshotWriter and SnapshotReader need to be interfaces because we will have a real implementation and a mocked implementation for testing. These two types are internal to raft and are not exposed to the state machine.

BatchedSnapshotReader and BatchedSnapshotWriter can depend on SnapshotWriter and SnapshotReade to reuse some code but this is not strictly required. These two type don't have to be interfaces if they delegate the IO to SnapshotWriter and SnapshotReader.

What do you think @hachikuji?

[hachikuji]

Ok, I think I see what you're saying. I guess I was not expecting that we would need a separate low-level interface when we already have FileRecords and MemoryRecords. I think Records already gives us a way to read from arbitrary positions:

    long writeTo(GatheringByteChannel channel, long position, int length) throws IOException;

But there is no Records implementation currently that allows unaligned writes. We only have this:

    public int append(MemoryRecords records) throws IOException;

Perhaps it would be possible to extend Records to provide what we need instead of creating a new interface?

As far as the naming, I wonder if we can reserve the nicer SnapshotXXX names for the state machine. Really what I would like is a common type that can be used by both handleSnapshot and handleCommit since both callbacks just need to provide a way to read through a set of records. I was thinking it could be BatchReader, but it doesn't have to be if there is something better.

(By the way, it's not too clear to me why we need freeze when we already have close.)

[jsancio]

Perhaps it would be possible to extend Records to provide what we need instead of creating a new interface?

I think you are right that we don't need SnapshotReader as Records provides all of the functionality we need. SnapshotReader was added so that we didn't expose all of the mutating APIs in FileRecords. Snapshot are supposed to be immutable once frozen.

I think we still want SnapshotWriter or something similar as it provides 1. raw writes of bytes and 2. freeze which optionally marks the snapshot as immutable if the validation passes.

For 1., I don't think we should add raw writes to Records or FileRecords as in essence we are exposing an unsafe API and the user needs to make sure that they are writing the correct data.

For 2., I think we can get away from introducing a new type/interface and instead add that functionality as a static method in Snapshots.java. Unittest (mock tests) maybe difficult with this code organization.

As far as the naming, I wonder if we can reserve the nicer SnapshotXXX names for the state machine.

Yes.I'll use the prettier name for the type exposed to the state machine.

Really what I would like is a common type that can be used by both handleSnapshot and handleCommit since both callbacks just need to provide a way to read through a set of records. I was thinking it could be BatchReader, but it doesn't have to be if there is something better.

I think this should be possible. I haven't implemented this part so I don't have all of the details. I think the requirement for the type sent through handleSnapshot are a subset of the requirements for the type sent through handleCommit. In particular, the Raft Client (ListenerContext) only needs to know when the snapshot has been read fully (e.g. close). The Raft Client doesn't need so to know what is the "last seen offset + 1" as it already knows the SnapshotId sent through handleSnapshot.

(By the way, it's not too clear to me why we need freeze when we already have close.)

I wanted to allow the user to abort a snapshot. If close is called without calling freeze then the partial snapshot is deleted and not made immutable.

[jsancio]

@hachikuji and I discussed this offline. The latest PR should reflect our discussion.

The interfaces RawSnapshotWriter and RawSnapshotReader are expected to change after we address https://issues.apache.org/jira/browse/KAFKA-10694

[jsancio]

Changes to this file are unrelated to this PR. Made them here so that docsJar would succeed.

[hachikuji]

Is there a tangible benefit to separating snapshots into a new directory? Currently the log directory is a flat structure. I'm wondering if we should stick with convention.

[jsancio]

I think this depends on if we need to scan the snapshot directory. Unfortunately, I don't have a concrete answer at the moment. When we implement the changes to the rest of the raft client. Log truncation, updating the start offset and LEO, we may need to scan the snapshot/checkpoint folder to determine the greatest log start offset and LEO. @lbradstreet suggested storing them in a different directory as part of the KIP-630 review process as Kafka already have a few files in the partition log directory.

[hachikuji]

Not sure if the snapshots will have much of an impact on this. It's not like we'll be storing more than a couple of them.

[hachikuji]

My preference is probably to keep things flat for now until we figure out what we want the long-term structure to look like. I guess it comes down to the implementation, but intuitively, the only time we'd need to scan would be on startup and we have to do that anyway.

[jsancio]

Done. Removed the snapshots directory.

[lbradstreet]

@jsancio @hachikuji we saw DeleteRecords calls became pretty expensive and tied up request handler threads when made on big partition directories and it scanned the directory to find deletable snapshot files. If we keep references to these snapshots in memory after log open this won't be a big deal to place them in the same directory. That's what we ended up doing for the producer state snapshots and it worked well.

[hachikuji]

Wonder if we should consider using Utils.atomicMoveWithFallback?

[jsancio]

I prefer to keep it this way. The atomic move (rename) should always succeed because we guarantee it is within the same directory. I think I prefer for Kafka to throw an exception than for the raft client or state machine to see a partial snapshot file because it performed a file copy instead of a file rename.

[hachikuji]

It might be useful to review the history behind Utils.atomicMoveWithFallback. It's not clear to me why this case is different from some of the other situations that it is used.

[jsancio]

I investigated the issue a bit. It look like this utility was introduce to replace these lines in OffsetCheckpoint.scala:

// swap new offset checkpoint file with previous one
if(!temp.renameTo(file)) {
    // renameTo() fails on Windows if the destination file exists.
    file.delete()
    if(!temp.renameTo(file))
        throw new IOException(...)
}

Looking at the JDK implementation in Windows, ATOMIC_MOVE should work in Windows if the target file exists: https://github.com/openjdk/jdk/blob/master/src/java.base/windows/classes/sun/nio/fs/WindowsFileCopy.java#L298-L312

In other words, I think we can keep the code as is in this PR.

cc @ijuma

[ijuma]

It's not only Windows, NFS also has some restrictions. Why don't you want to use the utility method?

[ijuma]

Also, I would not rely on reading the code to assume one way or another. You'd want to test it too.

[hachikuji]

LGTM. I am going to push a trivial tweak to make use of covariantCast in MockLog and to use the atomic move utility. We can open a separate issue if using the utility is a mistake and discuss further, but it seems reasonable to have the initial patch follow the established pattern.