MINOR: fix noticed typo in raft and metadata projects (apache#13612)

Reviewers: Josep Prat <jlprat@apache.org>

metadata/src/main/java/org/apache/kafka/controller/QuorumController.java

@@ -1585,7 +1585,7 @@ private void resetToEmptyState() {
     private int lastCommittedEpoch = -1;
 
     /**
-     * The timestamp in milliseconds of the last batch we have committed, or -1 if we have not commmitted any offset.
+     * The timestamp in milliseconds of the last batch we have committed, or -1 if we have not committed any offset.
      */
     private long lastCommittedTimestamp = -1;
 

metadata/src/main/java/org/apache/kafka/image/writer/ImageWriter.java

@@ -48,7 +48,7 @@ default void write(int version, ApiMessage message) {
     void write(ApiMessageAndVersion record);
 
     /**
-     * Close the image writer, dicarding all progress. Calling this function more than once has
+     * Close the image writer, discarding all progress. Calling this function more than once has
      * no effect.
      */
     default void close() {

raft/src/main/java/org/apache/kafka/raft/Batch.java

@@ -156,7 +156,7 @@ public int hashCode() {
      * @param epoch epoch of the leader that created this batch
      * @param appendTimestamp timestamp in milliseconds of when the batch was appended
      * @param sizeInBytes number of bytes used by this batch
-     * @param lastOffset offset of the last record of this batch
+     * @param records the list of records in this batch
      */
     public static <T> Batch<T> control(
         long baseOffset,

raft/src/main/java/org/apache/kafka/raft/CandidateState.java

@@ -39,7 +39,7 @@ public class CandidateState implements EpochState {
     private final Logger log;
 
     /**
-     * The life time of a candidate state is the following:
+     * The lifetime of a candidate state is the following:
      *
      *  1. Once started, it would keep record of the received votes.
      *  2. If majority votes granted, it can then end its life and will be replaced by a leader state;

raft/src/main/java/org/apache/kafka/raft/ExpirationService.java

@@ -25,7 +25,7 @@ public interface ExpirationService {
      * the provided time limit expires.
      *
      * @param timeoutMs the duration in milliseconds before the future is completed exceptionally
-     * @param <T> arbitrary future type (the service must set no expectation on the this type)
+     * @param <T> arbitrary future type (the service must set no expectation on this type)
      * @return the completable future
      */
     <T> CompletableFuture<T> failAfter(long timeoutMs);

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java

@@ -396,7 +396,7 @@ public void register(Listener<T> listener) {
     @Override
     public void unregister(Listener<T> listener) {
         pendingRegistrations.add(Registration.unregister(listener));
-        // No need to wakeup the polling thread. It is a removal so the updates can be
+        // No need to wake up the polling thread. It is a removal so the updates can be
         // delayed until the polling thread wakes up for other reasons.
     }
 

raft/src/main/java/org/apache/kafka/raft/RaftClient.java

@@ -105,7 +105,7 @@ default void beginShutdown() {}
     /**
      * Unregisters a listener.
      *
-     * To distinguish from events that happend before the call to {@code unregister} and a future
+     * To distinguish from events that happened before the call to {@code unregister} and a future
      * call to {@code register}, different {@code Listener} instances must be used.
      *
      * If the {@code Listener} provided was never registered then the unregistration is ignored. 
@@ -115,7 +115,7 @@ default void beginShutdown() {}
     void unregister(Listener<T> listener);
 
     /**
-     * Returns the current high water mark, or OptionalLong.empty if it is not known.
+     * Returns the current high watermark, or OptionalLong.empty if it is not known.
      */
     OptionalLong highWatermark();
 
@@ -226,7 +226,7 @@ default void beginShutdown() {}
      * @param snapshotId The ID of the new snapshot, which includes the (exclusive) last committed offset
      *                   and the last committed epoch.
      * @param lastContainedLogTime The append time of the highest record contained in this snapshot
-     * @return a writable snapshot if it doesn't already exists
+     * @return a writable snapshot if it doesn't already exist
      * @throws IllegalArgumentException if the committed offset is greater than the high-watermark
      *         or less than the log start offset.
      */

raft/src/main/java/org/apache/kafka/raft/ReplicatedLog.java

@@ -259,7 +259,7 @@ default long truncateToEndOffset(OffsetAndEpoch endOffset) {
      * the quorum leader.
      *
      * @param snapshotId the end offset and epoch that identifies the snapshot
-     * @return a writable snapshot if it doesn't already exists
+     * @return a writable snapshot if it doesn't already exist
      */
     Optional<RawSnapshotWriter> storeSnapshot(OffsetAndEpoch snapshotId);
 
@@ -287,7 +287,7 @@ default long truncateToEndOffset(OffsetAndEpoch endOffset) {
      /**
       * Returns the latest snapshot id if one exists.
       *
-      * @return an Optional snapshot id of the latest snashot if one exists, otherwise returns an
+      * @return an Optional snapshot id of the latest snapshot if one exists, otherwise returns an
       *         empty Optional
       */
     Optional<OffsetAndEpoch> latestSnapshotId();

raft/src/main/java/org/apache/kafka/snapshot/RawSnapshotReader.java

@@ -35,7 +35,7 @@ public interface RawSnapshotReader {
     long sizeInBytes();
 
     /**
-     * Creates a slize of unaligned records from the position up to a size.
+     * Creates a slice of unaligned records from the position up to a size.
      *
      * @param position the starting position of the slice in the snapshot
      * @param size the maximum size of the slice

raft/src/main/java/org/apache/kafka/snapshot/Snapshots.java

@@ -72,7 +72,7 @@ public static Path createTempFile(Path logDir, OffsetAndEpoch snapshotId) {
         Path dir = snapshotDir(logDir);
 
         try {
-            // Create the snapshot directory if it doesn't exists
+            // Create the snapshot directory if it doesn't exist
             Files.createDirectories(dir);
             String prefix = String.format("%s-", filenameFromSnapshotId(snapshotId));
             return Files.createTempFile(dir, prefix, PARTIAL_SUFFIX);