KAFKA-6157; Fix repeated words words in JavaDoc and comments.

efeg · becketqin · commit 86062e9a78dc · 2017-11-05T18:00:43.000-08:00
Author: Adem Efe Gencer &lt;agencer@linkedin.com&gt;

Reviewers: Jiangjie Qin &lt;becket.qin@gmail.com&gt;

Closes #4170 from efeg/bug/typoFix
diff --git a/clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java b/clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java
@@ -60,7 +60,7 @@ private Deque<NetworkClient.InFlightRequest> requestQueue(String node) {
     }
 
     /**
-     * Get the oldest request (the one that that will be completed next) for the given node
+     * Get the oldest request (the one that will be completed next) for the given node
      */
     public NetworkClient.InFlightRequest completeNext(String node) {
         return requestQueue(node).pollLast();
@@ -167,5 +167,5 @@ public List<String> getNodesWithTimedOutRequests(long now, int requestTimeoutMs)
         }
         return nodeIds;
     }
-    
+
 }
diff --git a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
@@ -369,7 +369,7 @@ private void doSend(ClientRequest clientRequest, boolean isInternalRequest, long
         if (!isInternalRequest) {
             // If this request came from outside the NetworkClient, validate
             // that we can send data.  If the request is internal, we trust
-            // that that internal code has done this validation.  Validation
+            // that internal code has done this validation.  Validation
             // will be slightly different for some internal requests (for
             // example, ApiVersionsRequests can be sent prior to being in
             // READY state.)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
@@ -423,7 +423,7 @@
  * <p>
  * Transactions were introduced in Kafka 0.11.0 wherein applications can write to multiple topics and partitions atomically.
  * In order for this to work, consumers reading from these partitions should be configured to only read committed data.
- * This can be achieved by by setting the {@code isolation.level=read_committed} in the consumer's configuration.
+ * This can be achieved by setting the {@code isolation.level=read_committed} in the consumer's configuration.
  *
  * <p>
  * In <code>read_committed</code> mode, the consumer will read only those transactional messages which have been
@@ -704,9 +704,9 @@ private KafkaConsumer(ConsumerConfig config,
             IsolationLevel isolationLevel = IsolationLevel.valueOf(
                     config.getString(ConsumerConfig.ISOLATION_LEVEL_CONFIG).toUpperCase(Locale.ROOT));
             Sensor throttleTimeSensor = Fetcher.throttleTimeSensor(metrics, metricsRegistry.fetcherMetrics);
-            
-            int heartbeatIntervalMs = config.getInt(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG); 
-            
+
+            int heartbeatIntervalMs = config.getInt(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG);
+
             NetworkClient netClient = new NetworkClient(
                     new Selector(config.getLong(ConsumerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG), metrics, time, metricGrpPrefix, channelBuilder, logContext),
                     this.metadata,
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
@@ -951,7 +951,7 @@ private void ensureValidRecordSize(int size) {
      * </p>
      * <p>
      * Applications don't need to call this method for transactional producers, since the {@link #commitTransaction()} will
-     * flush all buffered records before performing the commit. This ensures that all the the {@link #send(ProducerRecord)}
+     * flush all buffered records before performing the commit. This ensures that all the {@link #send(ProducerRecord)}
      * calls made since the previous {@link #beginTransaction()} are completed before the commit.
      * </p>
      *
diff --git a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java
@@ -384,7 +384,7 @@ private int writeLegacyCompressedWrapperHeader() {
     }
 
     /**
-     * Append a record and return its checksum for message format v0 and v1, or null for for v2 and above.
+     * Append a record and return its checksum for message format v0 and v1, or null for v2 and above.
      */
     private Long appendWithOffset(long offset, boolean isControlRecord, long timestamp, ByteBuffer key,
                                   ByteBuffer value, Header[] headers) {
diff --git a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java
@@ -186,7 +186,7 @@ public void authenticate() throws IOException {
                     if (authenticateVersion != null)
                         saslAuthenticateVersion((short) Math.min(authenticateVersion.maxVersion, ApiKeys.SASL_AUTHENTICATE.latestVersion()));
                     setSaslState(SaslState.SEND_HANDSHAKE_REQUEST);
-                    // Fall through to send send handshake request with the latest supported version
+                    // Fall through to send handshake request with the latest supported version
                 }
             case SEND_HANDSHAKE_REQUEST:
                 SaslHandshakeRequest handshakeRequest = createSaslHandshakeRequest(saslHandshakeVersion);
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java
@@ -1213,7 +1213,7 @@ public void onAssigned(ConnectProtocol.Assignment assignment, int generation) {
         public void onRevoked(String leader, Collection<String> connectors, Collection<ConnectorTaskId> tasks) {
             log.info("Rebalance started");
 
-            // Note that since we don't reset the assignment, we we don't revoke leadership here. During a rebalance,
+            // Note that since we don't reset the assignment, we don't revoke leadership here. During a rebalance,
             // it is still important to have a leader that can write configs, offsets, etc.
 
             if (rebalanceResolved) {
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java b/connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java
@@ -316,7 +316,7 @@ public void run() {
 
                     synchronized (KafkaBasedLog.this) {
                         // Only invoke exactly the number of callbacks we found before triggering the read to log end
-                        // since it is possible for another write + readToEnd to sneak in in the meantime
+                        // since it is possible for another write + readToEnd to sneak in the meantime
                         for (int i = 0; i < numCallbacks; i++) {
                             Callback<Void> cb = readLogEndOffsetCallbacks.poll();
                             cb.onCompletion(null, null);
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaConfigBackingStoreTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaConfigBackingStoreTest.java
@@ -282,7 +282,7 @@ public void testPutTaskConfigs() throws Exception {
         assertNull(configState.taskConfig(TASK_IDS.get(0)));
         assertNull(configState.taskConfig(TASK_IDS.get(1)));
 
-        // Writing task task configs should block until all the writes have been performed and the root record update
+        // Writing task configs should block until all the writes have been performed and the root record update
         // has completed
         List<Map<String, String>> taskConfigs = Arrays.asList(SAMPLE_CONFIGS.get(0), SAMPLE_CONFIGS.get(1));
         configStorage.putTaskConfigs("connector1", taskConfigs);
@@ -335,7 +335,7 @@ public void testPutTaskConfigsZeroTasks() throws Exception {
         ClusterConfigState configState = configStorage.snapshot();
         assertEquals(-1, configState.offset());
 
-        // Writing task task configs should block until all the writes have been performed and the root record update
+        // Writing task configs should block until all the writes have been performed and the root record update
         // has completed
         List<Map<String, String>> taskConfigs = Collections.emptyList();
         configStorage.putTaskConfigs("connector1", taskConfigs);
diff --git a/core/src/main/scala/kafka/cluster/Partition.scala b/core/src/main/scala/kafka/cluster/Partition.scala
@@ -325,7 +325,7 @@ class Partition(val topic: String,
   }
 
   /**
-   * Update the the follower's state in the leader based on the last fetch request. See
+   * Update the follower's state in the leader based on the last fetch request. See
    * [[kafka.cluster.Replica#updateLogReadResult]] for details.
    *
    * @return true if the leader's log start offset or high watermark have been updated
diff --git a/core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala b/core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala
@@ -380,7 +380,7 @@ private[group] class GroupMetadata(val groupId: String, initialState: GroupState
         pendingOffsets.foreach { case (topicPartition, commitRecordMetadataAndOffset) =>
           if (commitRecordMetadataAndOffset.appendedBatchOffset.isEmpty)
             throw new IllegalStateException(s"Trying to complete a transactional offset commit for producerId $producerId " +
-              s"and groupId $groupId even though the the offset commit record itself hasn't been appended to the log.")
+              s"and groupId $groupId even though the offset commit record itself hasn't been appended to the log.")
 
           val currentOffsetOpt = offsets.get(topicPartition)
           if (currentOffsetOpt.forall(_.olderThan(commitRecordMetadataAndOffset))) {
@@ -405,6 +405,7 @@ private[group] class GroupMetadata(val groupId: String, initialState: GroupState
 
   def removeOffsets(topicPartitions: Seq[TopicPartition]): immutable.Map[TopicPartition, OffsetAndMetadata] = {
     topicPartitions.flatMap { topicPartition =>
+
       pendingOffsetCommits.remove(topicPartition)
       pendingTransactionalOffsetCommits.foreach { case (_, pendingOffsets) =>
         pendingOffsets.remove(topicPartition)
diff --git a/core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala b/core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala
@@ -601,7 +601,7 @@ class TransactionStateManager(brokerId: Int,
 
           val append: Boolean = metadata.inLock {
             if (epochAndMetadata.coordinatorEpoch != coordinatorEpoch) {
-              // the coordinator epoch has changed, reply to client immediately with with NOT_COORDINATOR
+              // the coordinator epoch has changed, reply to client immediately with NOT_COORDINATOR
               responseCallback(Errors.NOT_COORDINATOR)
               false
             } else {
diff --git a/core/src/main/scala/kafka/server/DelayedDeleteRecords.scala b/core/src/main/scala/kafka/server/DelayedDeleteRecords.scala
@@ -62,7 +62,7 @@ class DelayedDeleteRecords(delayMs: Long,
   /**
    * The delayed delete records operation can be completed if every partition specified in the request satisfied one of the following:
    *
-   * 1) There was an error while checking if all replicas have caught up to to the deleteRecordsOffset: set an error in response
+   * 1) There was an error while checking if all replicas have caught up to the deleteRecordsOffset: set an error in response
    * 2) The low watermark of the partition has caught up to the deleteRecordsOffset. set the low watermark in response
    *
    */
diff --git a/core/src/main/scala/kafka/tools/JmxTool.scala b/core/src/main/scala/kafka/tools/JmxTool.scala
@@ -67,7 +67,7 @@ object JmxTool extends Logging {
       .describedAs("format")
       .ofType(classOf[String])
     val jmxServiceUrlOpt =
-      parser.accepts("jmx-url", "The url to connect to to poll JMX data. See Oracle javadoc for JMXServiceURL for details.")
+      parser.accepts("jmx-url", "The url to connect to poll JMX data. See Oracle javadoc for JMXServiceURL for details.")
         .withRequiredArg
         .describedAs("service-url")
         .ofType(classOf[String])
diff --git a/core/src/test/scala/integration/kafka/api/ProducerBounceTest.scala b/core/src/test/scala/integration/kafka/api/ProducerBounceTest.scala
@@ -60,7 +60,7 @@ class ProducerBounceTest extends KafkaServerTestHarness {
   private val topic1 = "topic-1"
 
   /**
-   * With replication, producer should able able to find new leader after it detects broker failure
+   * With replication, producer should able to find new leader after it detects broker failure
    */
   @Ignore // To be re-enabled once we can make it less flaky (KAFKA-2837)
   @Test
diff --git a/core/src/test/scala/unit/kafka/admin/DeleteConsumerGroupTest.scala b/core/src/test/scala/unit/kafka/admin/DeleteConsumerGroupTest.scala
@@ -141,7 +141,7 @@ class DeleteConsumerGroupTest extends KafkaServerTestHarness {
     AdminUtils.deleteAllConsumerGroupInfoForTopicInZK(zkUtils, topicToDelete)
 
     TestUtils.waitUntilTrue(() => !groupTopicDirsForTopicToDelete.exists(groupTopicOffsetAndOwnerDirsExist),
-      "Consumer group info on deleted topic topic should be deleted by DeleteAllConsumerGroupInfoForTopicInZK")
+      "Consumer group info on deleted topic should be deleted by DeleteAllConsumerGroupInfoForTopicInZK")
     TestUtils.waitUntilTrue(() => groupTopicDirsForOtherTopic.forall(groupTopicOffsetAndOwnerDirsExist),
       "Consumer group info on unrelated topics should not be deleted by DeleteAllConsumerGroupInfoForTopicInZK")
   }
diff --git a/core/src/test/scala/unit/kafka/log/LogManagerTest.scala b/core/src/test/scala/unit/kafka/log/LogManagerTest.scala
@@ -211,7 +211,7 @@ class LogManagerTest {
       log.appendAsLeader(set, leaderEpoch = 0)
     }
     time.sleep(logManager.InitialTaskDelayMs)
-    assertTrue("Time based flush should have been triggered triggered", lastFlush != log.lastFlushTime)
+    assertTrue("Time based flush should have been triggered", lastFlush != log.lastFlushTime)
   }
 
   /**
diff --git a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
@@ -328,7 +328,7 @@ class SocketServerTest extends JUnitSuite {
       TestUtils.waitUntilTrue(() => openChannel.isEmpty, "Idle channel not closed")
       TestUtils.waitUntilTrue(() => openOrClosingChannel.isDefined, "Channel removed without processing staged receives")
 
-      // Create new connection with same id when when `channel1` is in Selector.closingChannels
+      // Create new connection with same id when `channel1` is in Selector.closingChannels
       // Check that new connection is closed and openOrClosingChannel still contains `channel1`
       connectAndWaitForConnectionRegister()
       TestUtils.waitUntilTrue(() => connectionCount == 1, "Failed to close channel")
diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
@@ -1431,7 +1431,7 @@ object TestUtils extends Logging {
 
   private def asBytes(string: String) = string.getBytes(StandardCharsets.UTF_8)
 
-  // Verifies that the record was intended to be committed by checking the the headers for an expected transaction status
+  // Verifies that the record was intended to be committed by checking the headers for an expected transaction status
   // If true, this will return the value as a string. It is expected that the record in question should have been created
   // by the `producerRecordWithExpectedTransactionStatus` method.
   def assertCommittedAndGetValue(record: ConsumerRecord[Array[Byte], Array[Byte]]) : String = {
diff --git a/docs/connect.html b/docs/connect.html
@@ -329,7 +329,7 @@ <h5><a id="connect_taskexample" href="#connect_taskexample">Task Example - Sourc
         }
     </pre>
 
-    <p>These are slightly simplified versions, but show that that these methods should be relatively simple and the only work they should perform is allocating or freeing resources. There are two points to note about this implementation. First, the <code>start()</code> method does not yet handle resuming from a previous offset, which will be addressed in a later section. Second, the <code>stop()</code> method is synchronized. This will be necessary because <code>SourceTasks</code> are given a dedicated thread which they can block indefinitely, so they need to be stopped with a call from a different thread in the Worker.</p>
+    <p>These are slightly simplified versions, but show that these methods should be relatively simple and the only work they should perform is allocating or freeing resources. There are two points to note about this implementation. First, the <code>start()</code> method does not yet handle resuming from a previous offset, which will be addressed in a later section. Second, the <code>stop()</code> method is synchronized. This will be necessary because <code>SourceTasks</code> are given a dedicated thread which they can block indefinitely, so they need to be stopped with a call from a different thread in the Worker.</p>
 
     <p>Next, we implement the main functionality of the task, the <code>poll()</code> method which gets events from the input system and returns a <code>List&lt;SourceRecord&gt;</code>:</p>
 
diff --git a/docs/documentation/streams/architecture.html b/docs/documentation/streams/architecture.html
@@ -15,5 +15,5 @@
  limitations under the License.
 -->
 
-<!-- should always link the the latest release's documentation -->
+<!-- should always link the latest release's documentation -->
 <!--#include virtual="../../streams/architecture.html" -->
diff --git a/docs/documentation/streams/core-concepts.html b/docs/documentation/streams/core-concepts.html
@@ -15,5 +15,5 @@
  limitations under the License.
 -->
 
-<!-- should always link the the latest release's documentation -->
+<!-- should always link the latest release's documentation -->
 <!--#include virtual="../../streams/core-concepts.html" -->
diff --git a/docs/documentation/streams/developer-guide.html b/docs/documentation/streams/developer-guide.html
@@ -15,5 +15,5 @@
  limitations under the License.
 -->
 
-<!-- should always link the the latest release's documentation -->
+<!-- should always link the latest release's documentation -->
 <!--#include virtual="../../streams/developer-guide.html" -->
diff --git a/docs/documentation/streams/index.html b/docs/documentation/streams/index.html
@@ -15,5 +15,5 @@
  limitations under the License.
 -->
 
-<!-- should always link the the latest release's documentation -->
+<!-- should always link the latest release's documentation -->
 <!--#include virtual="../../streams/index.html" -->
diff --git a/docs/documentation/streams/quickstart.html b/docs/documentation/streams/quickstart.html
@@ -15,5 +15,5 @@
  limitations under the License.
 -->
 
-<!-- should always link the the latest release's documentation -->
+<!-- should always link the latest release's documentation -->
 <!--#include virtual="../../streams/quickstart.html" -->
diff --git a/docs/documentation/streams/tutorial.html b/docs/documentation/streams/tutorial.html
@@ -15,5 +15,5 @@
  limitations under the License.
 -->
 
-<!-- should always link the the latest release's documentation -->
+<!-- should always link the latest release's documentation -->
 <!--#include virtual="../../streams/tutorial.html" -->
diff --git a/docs/documentation/streams/upgrade-guide.html b/docs/documentation/streams/upgrade-guide.html
@@ -15,5 +15,5 @@
  limitations under the License.
 -->
 
-<!-- should always link the the latest release's documentation -->
+<!-- should always link the latest release's documentation -->
 <!--#include virtual="../../streams/upgrade-guide.html" -->
diff --git a/docs/implementation.html b/docs/implementation.html
@@ -99,7 +99,7 @@ <h5><a id="recordheader" href="#recordheader">5.4.2.1 Record Header</a></h5>
 		headerValueLength: varint
 		Value: byte[]
 	</pre></p>
-    <p>We use the the same varint encoding as Protobuf. More information on the latter can be found <a href="https://developers.google.com/protocol-buffers/docs/encoding#varints">here</a>. The count of headers in a record
+    <p>We use the same varint encoding as Protobuf. More information on the latter can be found <a href="https://developers.google.com/protocol-buffers/docs/encoding#varints">here</a>. The count of headers in a record
     is also encoded as a varint.</p>
 
     <h3><a id="log" href="#log">5.4 Log</a></h3>
diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java b/streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
@@ -59,7 +59,7 @@ public class StreamsBuilder {
     final InternalTopologyBuilder internalTopologyBuilder = topology.internalTopologyBuilder;
 
     private final InternalStreamsBuilder internalStreamsBuilder = new InternalStreamsBuilder(internalTopologyBuilder);
-    
+
     /**
      * Create a {@link KStream} from the specified topics.
      * The default {@code "auto.offset.reset"} strategy, default {@link TimestampExtractor}, and default key and value
@@ -237,7 +237,7 @@ public synchronized <K, V> KTable<K, V> table(final String topic,
      * If this is not the case the returned {@link KTable} will be corrupted.
      * <p>
      * The resulting {@link KTable} will be materialized in a local {@link KeyValueStore} with an internal
-     * store name. Note that that store name may not be queriable through Interactive Queries.
+     * store name. Note that store name may not be queriable through Interactive Queries.
      * No internal changelog topic is created since the original input topic can be used for recovery (cf.
      * methods of {@link KGroupedStream} and {@link KGroupedTable} that return a {@link KTable}).
      *
@@ -258,7 +258,7 @@ public synchronized <K, V> KTable<K, V> table(final String topic) {
      * If this is not the case the returned {@link KTable} will be corrupted.
      * <p>
      * The resulting {@link KTable} will be materialized in a local {@link KeyValueStore} with an internal
-     * store name. Note that that store name may not be queriable through Interactive Queries.
+     * store name. Note that store name may not be queriable through Interactive Queries.
      * No internal changelog topic is created since the original input topic can be used for recovery (cf.
      * methods of {@link KGroupedStream} and {@link KGroupedTable} that return a {@link KTable}).
      *
@@ -312,7 +312,7 @@ public synchronized <K, V> KTable<K, V> table(final String topic,
      * Input {@link KeyValue records} with {@code null} key will be dropped.
      * <p>
      * The resulting {@link GlobalKTable} will be materialized in a local {@link KeyValueStore} with an internal
-     * store name. Note that that store name may not be queriable through Interactive Queries.
+     * store name. Note that store name may not be queriable through Interactive Queries.
      * No internal changelog topic is created since the original input topic can be used for recovery (cf.
      * methods of {@link KGroupedStream} and {@link KGroupedTable} that return a {@link KTable}).
      * <p>
@@ -343,7 +343,7 @@ public synchronized <K, V> GlobalKTable<K, V> globalTable(final String topic,
      * Input {@link KeyValue records} with {@code null} key will be dropped.
      * <p>
      * The resulting {@link GlobalKTable} will be materialized in a local {@link KeyValueStore} with an internal
-     * store name. Note that that store name may not be queriable through Interactive Queries.
+     * store name. Note that store name may not be queriable through Interactive Queries.
      * No internal changelog topic is created since the original input topic can be used for recovery (cf.
      * methods of {@link KGroupedStream} and {@link KGroupedTable} that return a {@link KTable}).
      * <p>
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/KStreamBuilder.java b/streams/src/main/java/org/apache/kafka/streams/kstream/KStreamBuilder.java
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamAggregate.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamAggregate.java
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamReduce.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamReduce.java
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregate.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregate.java
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamWindowReduce.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamWindowReduce.java
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableAggregate.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableAggregate.java
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableReduce.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableReduce.java
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
diff --git a/tests/setup.cfg b/tests/setup.cfg
diff --git a/tests/unit/setup.cfg b/tests/unit/setup.cfg
diff --git a/tools/src/main/java/org/apache/kafka/tools/ProducerPerformance.java b/tools/src/main/java/org/apache/kafka/tools/ProducerPerformance.java

Original file line number	Diff line number	Diff line change
`@@ -60,7 +60,7 @@ private Deque<NetworkClient.InFlightRequest> requestQueue(String node) {`
`60`	`60`	`}`
`61`	`61`
`62`	`62`	`/**`
`63`		`- * Get the oldest request (the one that that will be completed next) for the given node`
	`63`	`+ * Get the oldest request (the one that will be completed next) for the given node`
`64`	`64`	`*/`
`65`	`65`	`public NetworkClient.InFlightRequest completeNext(String node) {`
`66`	`66`	`return requestQueue(node).pollLast();`
`@@ -167,5 +167,5 @@ public List<String> getNodesWithTimedOutRequests(long now, int requestTimeoutMs)`
`167`	`167`	`}`
`168`	`168`	`return nodeIds;`
`169`	`169`	`}`
`170`		`-`
	`170`	`+`
`171`	`171`	`}`
Original file line number	Diff line number	Diff line change
`@@ -951,7 +951,7 @@ private void ensureValidRecordSize(int size) {`
`951`	`951`	`* </p>`
`952`	`952`	`* <p>`
`953`	`953`	`* Applications don't need to call this method for transactional producers, since the {@link #commitTransaction()} will`
`954`		`- * flush all buffered records before performing the commit. This ensures that all the the {@link #send(ProducerRecord)}`
	`954`	`+ * flush all buffered records before performing the commit. This ensures that all the {@link #send(ProducerRecord)}`
`955`	`955`	`* calls made since the previous {@link #beginTransaction()} are completed before the commit.`
`956`	`956`	`* </p>`
`957`	`957`	`*`
Original file line number	Diff line number	Diff line change
`@@ -384,7 +384,7 @@ private int writeLegacyCompressedWrapperHeader() {`
`384`	`384`	`}`
`385`	`385`
`386`	`386`	`/**`
`387`		`- * Append a record and return its checksum for message format v0 and v1, or null for for v2 and above.`
	`387`	`+ * Append a record and return its checksum for message format v0 and v1, or null for v2 and above.`
`388`	`388`	`*/`
`389`	`389`	`private Long appendWithOffset(long offset, boolean isControlRecord, long timestamp, ByteBuffer key,`
`390`	`390`	`ByteBuffer value, Header[] headers) {`
Original file line number	Diff line number	Diff line change
`@@ -186,7 +186,7 @@ public void authenticate() throws IOException {`
`186`	`186`	`if (authenticateVersion != null)`
`187`	`187`	`saslAuthenticateVersion((short) Math.min(authenticateVersion.maxVersion, ApiKeys.SASL_AUTHENTICATE.latestVersion()));`
`188`	`188`	`setSaslState(SaslState.SEND_HANDSHAKE_REQUEST);`
`189`		`- // Fall through to send send handshake request with the latest supported version`
	`189`	`+ // Fall through to send handshake request with the latest supported version`
`190`	`190`	`}`
`191`	`191`	`case SEND_HANDSHAKE_REQUEST:`
`192`	`192`	`SaslHandshakeRequest handshakeRequest = createSaslHandshakeRequest(saslHandshakeVersion);`
Original file line number	Diff line number	Diff line change
`@@ -325,7 +325,7 @@ class Partition(val topic: String,`
`325`	`325`	`}`
`326`	`326`
`327`	`327`	`/**`
`328`		`- * Update the the follower's state in the leader based on the last fetch request. See`
	`328`	`+ * Update the follower's state in the leader based on the last fetch request. See`
`329`	`329`	`* [[kafka.cluster.Replica#updateLogReadResult]] for details.`
`330`	`330`	`*`
`331`	`331`	`* @return true if the leader's log start offset or high watermark have been updated`
Original file line number	Diff line number	Diff line change
`@@ -62,7 +62,7 @@ class DelayedDeleteRecords(delayMs: Long,`
`62`	`62`	`/**`
`63`	`63`	`* The delayed delete records operation can be completed if every partition specified in the request satisfied one of the following:`
`64`	`64`	`*`
`65`		`- * 1) There was an error while checking if all replicas have caught up to to the deleteRecordsOffset: set an error in response`
	`65`	`+ * 1) There was an error while checking if all replicas have caught up to the deleteRecordsOffset: set an error in response`
`66`	`66`	`* 2) The low watermark of the partition has caught up to the deleteRecordsOffset. set the low watermark in response`
`67`	`67`	`*`
`68`	`68`	`*/`
Original file line number	Diff line number	Diff line change
`@@ -141,7 +141,7 @@ class DeleteConsumerGroupTest extends KafkaServerTestHarness {`
`141`	`141`	`AdminUtils.deleteAllConsumerGroupInfoForTopicInZK(zkUtils, topicToDelete)`
`142`	`142`
`143`	`143`	`TestUtils.waitUntilTrue(() => !groupTopicDirsForTopicToDelete.exists(groupTopicOffsetAndOwnerDirsExist),`
`144`		`- "Consumer group info on deleted topic topic should be deleted by DeleteAllConsumerGroupInfoForTopicInZK")`
	`144`	`+ "Consumer group info on deleted topic should be deleted by DeleteAllConsumerGroupInfoForTopicInZK")`
`145`	`145`	`TestUtils.waitUntilTrue(() => groupTopicDirsForOtherTopic.forall(groupTopicOffsetAndOwnerDirsExist),`
`146`	`146`	`"Consumer group info on unrelated topics should not be deleted by DeleteAllConsumerGroupInfoForTopicInZK")`
`147`	`147`	`}`
Original file line number	Diff line number	Diff line change
`@@ -211,7 +211,7 @@ class LogManagerTest {`
`211`	`211`	`log.appendAsLeader(set, leaderEpoch = 0)`
`212`	`212`	`}`
`213`	`213`	`time.sleep(logManager.InitialTaskDelayMs)`
`214`		`- assertTrue("Time based flush should have been triggered triggered", lastFlush != log.lastFlushTime)`
	`214`	`+ assertTrue("Time based flush should have been triggered", lastFlush != log.lastFlushTime)`
`215`	`215`	`}`
`216`	`216`
`217`	`217`	`/**`
Original file line number	Diff line number	Diff line change
`@@ -329,7 +329,7 @@ <h5><a id="connect_taskexample" href="#connect_taskexample">Task Example - Sourc`
`329`	`329`	`}`
`330`	`330`	`</pre>`
`331`	`331`
`332`		- <p>These are slightly simplified versions, but show that that these methods should be relatively simple and the only work they should perform is allocating or freeing resources. There are two points to note about this implementation. First, the <code>start()</code> method does not yet handle resuming from a previous offset, which will be addressed in a later section. Second, the <code>stop()</code> method is synchronized. This will be necessary because <code>SourceTasks</code> are given a dedicated thread which they can block indefinitely, so they need to be stopped with a call from a different thread in the Worker.</p>
	`332`	+ <p>These are slightly simplified versions, but show that these methods should be relatively simple and the only work they should perform is allocating or freeing resources. There are two points to note about this implementation. First, the <code>start()</code> method does not yet handle resuming from a previous offset, which will be addressed in a later section. Second, the <code>stop()</code> method is synchronized. This will be necessary because <code>SourceTasks</code> are given a dedicated thread which they can block indefinitely, so they need to be stopped with a call from a different thread in the Worker.</p>
`333`	`333`
`334`	`334`	`<p>Next, we implement the main functionality of the task, the <code>poll()</code> method which gets events from the input system and returns a <code>List<SourceRecord></code>:</p>`
`335`	`335`