Fix typos in multiple files (#19086)

Fix typos in multiple files

Reviewers: Andrew Schofield <aschofield@confluent.io>

NOTICE-binary

@@ -349,7 +349,7 @@ The project maintains the following source code repositories:
 Angular JS, v1.6.6
 * License MIT (http://www.opensource.org/licenses/mit-license.php)
 * Project: http://angularjs.org
-* Coyright: (c) 2010-2017 Google, Inc.
+* Copyright: (c) 2010-2017 Google, Inc.
 
 aopalliance Version 1
 * License: all the source code provided by AOP Alliance is Public Domain.

clients/src/main/java/org/apache/kafka/common/KafkaFuture.java

@@ -135,7 +135,7 @@ public abstract class KafkaFuture<T> implements Future<T> {
      * The action may be invoked by the thread that calls {@code whenComplete} or it may be invoked by the thread that
      * completes the future.
      *
-     * @param action the action to preform
+     * @param action the action to perform
      * @return the new future
      */
     public abstract KafkaFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action);

clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java

@@ -304,7 +304,7 @@ public class KafkaConsumerTest {
 
             KafkaMetric existingMetric = (KafkaMetric) consumer.metrics().entrySet().iterator().next().getValue();
             consumer.registerMetricForSubscription(existingMetric);
-            // This test would fail without the check as the exising metric is registered in the consumer on startup
+            // This test would fail without the check as the existing metric is registered in the consumer on startup
             Mockito.verify(clientTelemetryReporter, atMostOnce()).metricChange(existingMetric);
         }
     }

clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java

@@ -2728,7 +2728,7 @@ public class KafkaProducerTest {
                     props, new StringSerializer(), new StringSerializer())) {
                 KafkaMetric existingMetric = (KafkaMetric) producer.metrics().entrySet().iterator().next().getValue();
                 producer.registerMetricForSubscription(existingMetric);
-                // This test would fail without the check as the exising metric is registered in the producer on startup
+                // This test would fail without the check as the existing metric is registered in the producer on startup
                 Mockito.verify(clientTelemetryReporter, atMostOnce()).metricChange(existingMetric);
             }
         }
@@ -2747,7 +2747,7 @@ public class KafkaProducerTest {
                     props, new StringSerializer(), new StringSerializer())) {
                 KafkaMetric existingMetric = (KafkaMetric) producer.metrics().entrySet().iterator().next().getValue();
                 producer.unregisterMetricFromSubscription(existingMetric);
-                // This test would fail without the check as the exising metric is registered in the consumer on startup
+                // This test would fail without the check as the existing metric is registered in the consumer on startup
                 Mockito.verify(clientTelemetryReporter, never()).metricRemoval(existingMetric);
             }
         }

connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java

@@ -1258,7 +1258,7 @@ public final class KafkaConfigBackingStore extends KafkaTopicBasedBackingStore i
         } else {
             // TRACE level since there may be many of these records in the config topic
             log.trace(
-                    "Ignoring old logging level {} for namespace {} that was writen to the config topic before this worker completed startup",
+                    "Ignoring old logging level {} for namespace {} that was written to the config topic before this worker completed startup",
                     level,
                     namespace
             );

connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedConnect.java

@@ -495,7 +495,7 @@ abstract class EmbeddedConnect {
         Response response = requestPost(restartEndpoint, "", Collections.emptyMap());
         try {
             if (response.getStatus() < Response.Status.BAD_REQUEST.getStatusCode()) {
-                //only the 202 stauts returns a body
+                //only the 202 status returns a body
                 if (response.getStatus() == Response.Status.ACCEPTED.getStatusCode()) {
                     return mapper.readerFor(ConnectorStateInfo.class)
                             .readValue(responseToString(response));

coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java

@@ -493,7 +493,7 @@ public class CoordinatorRuntime<S extends CoordinatorShard<U>, U> implements Aut
         final MemoryRecordsBuilder builder;
 
         /**
-         * The timer used to enfore the append linger time if
+         * The timer used to enforce the append linger time if
          * it is non-zero.
          */
         final Optional<TimerTask> lingerTimeoutTask;

core/src/main/scala/kafka/server/FetchSession.scala

@@ -224,7 +224,7 @@ class CachedPartition(var topic: String,
   * fields are read or modified.  This includes modification of the session partition map.
   *
   * @param id                 The unique fetch session ID.
-  * @param privileged         True if this session is privileged.  Sessions crated by followers
+  * @param privileged         True if this session is privileged.  Sessions created by followers
   *                           are privileged; session created by consumers are not.
   * @param partitionMap       The CachedPartitionMap.
   * @param usesTopicIds       True if this session is using topic IDs

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

@@ -2734,7 +2734,7 @@ public class SharePartitionManagerTest {
 
         SharePartition sp1 = mock(SharePartition.class);
         // Do not make the share partition acquirable hence it shouldn't be removed from the cache,
-        // as it won't be part of replica manger readFromLog request.
+        // as it won't be part of replica manager readFromLog request.
         when(sp1.maybeAcquireFetchLock()).thenReturn(false);
         when(sp1.maybeInitialize()).thenReturn(CompletableFuture.completedFuture(null));