MINOR: Fix various typos and formatting issues across multiple modules (#22177)

This is done via a script I wrote to report typos based on statistical
analysis and after I checked them manually, I applied the correct ones.
No AI was used.

I know it's not a very useful PR but I hope it's fine. Thank you!

Reviewers: Murali Basani <muralidhar.basani@aiven.io>, Pratyaksh Sharma
 <pratyaksh13@gmail.com>, Ken Huang <s7133700@gmail.com>, Chia-Ping Tsai
 <chia7712@gmail.com>

Author: Ladsgroup

clients/src/test/java/org/apache/kafka/clients/FetchSessionHandlerTest.java

@@ -124,10 +124,10 @@ private static void assertMapEquals(Map<TopicPartition, FetchRequest.PartitionDa
             if (!actualIter.hasNext()) {
                 fail("Element " + i + " not found.");
             }
-            Map.Entry<TopicPartition, FetchRequest.PartitionData> actuaLEntry = actualIter.next();
-            assertEquals(expectedEntry.getKey(), actuaLEntry.getKey(), "Element " + i +
+            Map.Entry<TopicPartition, FetchRequest.PartitionData> actualEntry = actualIter.next();
+            assertEquals(expectedEntry.getKey(), actualEntry.getKey(), "Element " + i +
                 " had a different TopicPartition than expected.");
-            assertEquals(expectedEntry.getValue(), actuaLEntry.getValue(), "Element " + i +
+            assertEquals(expectedEntry.getValue(), actualEntry.getValue(), "Element " + i +
                 " had different PartitionData than expected.");
             i++;
         }

clients/src/test/java/org/apache/kafka/clients/consumer/internals/CompletedFetchTest.java

@@ -92,7 +92,7 @@ record = records.get(0);
     @Test
     public void testAbortedTransactionRecordsRemoved() {
         int numRecords = 10;
-        Records rawRecords = newTranscactionalRecords(ControlRecordType.ABORT, numRecords);
+        Records rawRecords = newTransactionalRecords(ControlRecordType.ABORT, numRecords);
 
         FetchResponseData.PartitionData partitionData = new FetchResponseData.PartitionData()
                 .setRecords(rawRecords)
@@ -114,7 +114,7 @@ public void testAbortedTransactionRecordsRemoved() {
     @Test
     public void testCommittedTransactionRecordsIncluded() {
         int numRecords = 10;
-        Records rawRecords = newTranscactionalRecords(ControlRecordType.COMMIT, numRecords);
+        Records rawRecords = newTransactionalRecords(ControlRecordType.COMMIT, numRecords);
         FetchResponseData.PartitionData partitionData = new FetchResponseData.PartitionData()
                 .setRecords(rawRecords);
         CompletedFetch completedFetch = newCompletedFetch(0, partitionData);
@@ -258,7 +258,7 @@ private Records newRecords(long baseOffset, int count, long firstMessageId) {
         }
     }
 
-    private Records newTranscactionalRecords(ControlRecordType controlRecordType, int numRecords) {
+    private Records newTransactionalRecords(ControlRecordType controlRecordType, int numRecords) {
         Time time = new MockTime();
         ByteBuffer buffer = ByteBuffer.allocate(1024);
 

core/src/test/java/kafka/server/share/SharePartitionTest.java

@@ -5278,7 +5278,7 @@ public void testReleaseAcquiredRecordsWithAnotherMemberAndSubsetAcknowledged() {
         // Ack subset of records by "member-2".
         sharePartition.acknowledge("member-2",
                 List.of(new ShareAcknowledgementBatch(5, 5, List.of(AcknowledgeType.ACCEPT.id))));
-        // After the acknowledgements, the startOffset will be upadated to 6, since offset 5 is Terminal. Hence
+        // After the acknowledgements, the startOffset will be updated to 6, since offset 5 is Terminal. Hence
         // deliveryCompleteCount will remain 9.
         assertEquals(9, sharePartition.deliveryCompleteCount());
 
@@ -5465,7 +5465,7 @@ public void testMaxDeliveryCountLimitExceededForRecordsSubsetCacheCleared() {
         )));
 
         // After acknowledgements, since offsets 10 -> 12 are at the start of the caches state and are in Terminal state,
-        // the start offset will be updated to 13. From the remaining offstes in flight, only records (17 -> 19) are in Terminal state.
+        // the start offset will be updated to 13. From the remaining offsets in flight, only records (17 -> 19) are in Terminal state.
         assertEquals(3, sharePartition.deliveryCompleteCount());
 
         // Send next batch from offset 13, only 2 records should be acquired.
@@ -9709,7 +9709,7 @@ public void testAcquisitionLockTimeoutWithConcurrentAcknowledgement() throws Int
         TimerTask timerTask2 = sharePartition.cachedState().get(5L).batchAcquisitionLockTimeoutTask();
 
         // Acknowledge 1 offset in first batch as Accept to create offset tracking, accept complete
-        // sencond batch. And mark offset 0 as release so cached state do not move ahead.
+        // second batch. And mark offset 0 as release so cached state do not move ahead.
         sharePartition.acknowledge(MEMBER_ID, List.of(
             new ShareAcknowledgementBatch(0, 0, List.of(AcknowledgeType.RELEASE.id)),
             new ShareAcknowledgementBatch(1, 1, List.of(AcknowledgeType.ACCEPT.id)),

group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java

@@ -5805,7 +5805,7 @@ public void testStaticMembersJoinGroupWithMaxSize() {
         assertEquals(groupMaxSize, group.numAwaitingJoinResponse());
         assertTrue(group.isInState(PREPARING_REBALANCE));
 
-        // Advance clock by group initial rebalance delay to complete first inital delayed join.
+        // Advance clock by group initial rebalance delay to complete first initial delayed join.
         // This will extend the initial rebalance as new members have joined.
         GroupMetadataManagerTestContext.assertNoOrEmptyResult(context.sleep(50));
         // Advance clock by group initial rebalance delay to complete second inital delayed join.

group-coordinator/src/test/java/org/apache/kafka/coordinator/group/modern/TargetAssignmentBuilderTest.java

@@ -217,8 +217,8 @@ private MemberSubscriptionAndAssignmentImpl newMemberSubscriptionAndAssignment(
         }
 
         public TargetAssignmentBuilder.TargetAssignmentResult build() {
-            CoordinatorMetadataImage cooridnatorMetadataImage = new KRaftCoordinatorMetadataImage(metadataImageBuilder.build());
-            TopicIds.TopicResolver topicResolver = new TopicIds.CachedTopicResolver(cooridnatorMetadataImage);
+            CoordinatorMetadataImage coordinatorMetadataImage = new KRaftCoordinatorMetadataImage(metadataImageBuilder.build());
+            TopicIds.TopicResolver topicResolver = new TopicIds.CachedTopicResolver(coordinatorMetadataImage);
             // Prepare expected member specs.
             Map<String, MemberSubscriptionAndAssignmentImpl> memberSubscriptions = new HashMap<>();
 
@@ -256,7 +256,7 @@ public TargetAssignmentBuilder.TargetAssignmentResult build() {
             });
 
             // Prepare the expected subscription topic metadata.
-            SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(cooridnatorMetadataImage);
+            SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(coordinatorMetadataImage);
             SubscriptionType subscriptionType = HOMOGENEOUS;
 
             // Prepare the member assignments per topic partition.
@@ -284,7 +284,7 @@ public TargetAssignmentBuilder.TargetAssignmentResult build() {
                     .withSubscriptionType(subscriptionType)
                     .withTargetAssignment(targetAssignment)
                     .withInvertedTargetAssignment(invertedTargetAssignment)
-                    .withMetadataImage(cooridnatorMetadataImage)
+                    .withMetadataImage(coordinatorMetadataImage)
                     .withResolvedRegularExpressions(resolvedRegularExpressions);
 
             // Add the updated members or delete the deleted members.

raft/src/main/java/org/apache/kafka/raft/internals/KafkaRaftLog.java

@@ -563,10 +563,10 @@ private boolean cleanSnapshots(Function<OffsetAndEpoch, Optional<SnapshotDeletio
         }
 
         boolean didClean = false;
-        List<OffsetAndEpoch> epoches = new ArrayList<>(snapshots.keySet());
-        for (int i = 0; i < epoches.size() - 1; i++) {
-            OffsetAndEpoch epoch = epoches.get(i);
-            OffsetAndEpoch nextEpoch = epoches.get(i + 1);
+        List<OffsetAndEpoch> epochs = new ArrayList<>(snapshots.keySet());
+        for (int i = 0; i < epochs.size() - 1; i++) {
+            OffsetAndEpoch epoch = epochs.get(i);
+            OffsetAndEpoch nextEpoch = epochs.get(i + 1);
             Optional<SnapshotDeletionReason> reason = predicate.apply(epoch);
             if (reason.isPresent()) {
                 boolean deleted = deleteBeforeSnapshot(nextEpoch, reason.get());

streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java

@@ -1823,7 +1823,7 @@ public <V> void shouldHandleWindowKeyQuery(
         final Integer key,
         final Instant timeFrom,
         final Instant timeTo,
-        final Function<V, Integer> valueExtactor,
+        final Function<V, Integer> valueExtractor,
         final Set<Integer> expectedValues) {
 
         final WindowKeyQuery<Integer, V> query = WindowKeyQuery.withKeyAndWindowStartRange(
@@ -1864,7 +1864,7 @@ public <V> void shouldHandleWindowKeyQuery(
 
                 try (final WindowStoreIterator<V> iterator = queryResult.get(partition).getResult()) {
                     while (iterator.hasNext()) {
-                        actualValues.add(valueExtactor.apply(iterator.next().value));
+                        actualValues.add(valueExtractor.apply(iterator.next().value));
                     }
                 }
                 assertThat(queryResult.get(partition).getExecutionInfo(), is(empty()));
@@ -1877,7 +1877,7 @@ public <V> void shouldHandleWindowKeyQuery(
     public <V> void shouldHandleWindowRangeQuery(
         final Instant timeFrom,
         final Instant timeTo,
-        final Function<V, Integer> valueExtactor,
+        final Function<V, Integer> valueExtractor,
         final Set<Integer> expectedValues) {
 
         final WindowRangeQuery<Integer, V> query = WindowRangeQuery.withWindowStartRange(timeFrom, timeTo);
@@ -1914,7 +1914,7 @@ public <V> void shouldHandleWindowRangeQuery(
 
                 try (final KeyValueIterator<Windowed<Integer>, V> iterator = queryResult.get(partition).getResult()) {
                     while (iterator.hasNext()) {
-                        actualValues.add(valueExtactor.apply(iterator.next().value));
+                        actualValues.add(valueExtractor.apply(iterator.next().value));
                     }
                 }
                 assertThat(queryResult.get(partition).getExecutionInfo(), is(empty()));
@@ -2057,4 +2057,4 @@ private static Properties streamsConfiguration(final boolean cache, final boolea
         StreamsTestUtils.maybeSetDslStoreFormatHeaders(config, withHeaders);
         return config;
     }
-}
+}

streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KafkaStreamsCloseOptionsIntegrationTest.java

@@ -115,7 +115,7 @@ public void before(final TestInfo testName) throws Exception {
         streamsConfig.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 100);
         streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
         // In this test, we set the SESSION_TIMEOUT_MS_CONFIG high in order to show that the call to
-        // `close(CloseOptions)` can remove the application from the Consumder Groups successfully.
+        // `close(CloseOptions)` can remove the application from the Consumer Groups successfully.
         streamsConfig.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, Integer.MAX_VALUE);
         streamsConfig.putAll(commonClientConfig);
 

streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/SuppressionDurabilityIntegrationTest.java

@@ -158,7 +158,7 @@ public void shouldRecoverBufferAfterShutdown(final boolean withHeaders, final Te
         try {
             // start by putting some stuff in the buffer
             // note, we send all input records to partition 0
-            // to make sure that supppress doesn't erroneously send records to other partitions.
+            // to make sure that suppress doesn't erroneously send records to other partitions.
             produceSynchronouslyToPartitionZero(
                 input,
                 asList(
@@ -316,4 +316,4 @@ private static void produceSynchronouslyToPartitionZero(final String topic, fina
         ));
         IntegrationTestUtils.produceSynchronously(producerConfig, false, topic, Optional.of(0), toProduce);
     }
-}
+}

streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java

@@ -757,7 +757,7 @@ <KOut> KGroupedStream<KOut, V> groupBy(final KeyValueMapper<? super K, ? super V
      * <p>You can retrieve all generated internal topic names via {@link Topology#describe()}.
      * To explicitly set key/value serdes, to customize the names of the repartition and changelog topic, or to
      * customize the used state store, use {@link #join(KStream, ValueJoiner, JoinWindows, StreamJoined)}.
-     * For more control over the repartitioning, use {@link #repartition(Repartitioned)} on eiter input before {@code join()}.
+     * For more control over the repartitioning, use {@link #repartition(Repartitioned)} on either input before {@code join()}.
      *
      * @param rightStream
      *        the {@code KStream} to be joined with this stream
@@ -1592,4 +1592,4 @@ <VOut> KStream<K, VOut> processValues(
         final Named named,
         final String... stateStoreNames
     );
-}
+}