MINOR: Small javadoc/code cleanups in connect api and transforms (#12558)

Reviewers: Luke Chen <showuon@gmail.com>, Divij Vaidya <diviv@amazon.com>

connect/api/src/main/java/org/apache/kafka/connect/components/Versioned.java

@@ -24,7 +24,7 @@ public interface Versioned {
     /**
      * Get the version of this component.
      *
-     * @return the version, formatted as a String. The version may not be (@code null} or empty.
+     * @return the version, formatted as a String. The version may not be {@code null} or empty.
      */
     String version();
 }

connect/api/src/main/java/org/apache/kafka/connect/connector/policy/ConnectorClientConfigOverridePolicy.java

@@ -34,7 +34,7 @@ public interface ConnectorClientConfigOverridePolicy extends Configurable, AutoC
     /**
      * Worker will invoke this while constructing the producer for the SourceConnectors,  DLQ for SinkConnectors and the consumer for the
      * SinkConnectors to validate if all of the overridden client configurations are allowed per the
-     * policy implementation. This would also be invoked during the validate of connector configs via the Rest API.
+     * policy implementation. This would also be invoked during the validation of connector configs via the Rest API.
      *
      * If there are any policy violations, the connector will not be started.
      *

connect/api/src/main/java/org/apache/kafka/connect/data/Schema.java

@@ -104,7 +104,7 @@ public interface Schema {
          */
         STRUCT;
 
-        private String name;
+        private final String name;
 
         Type() {
             this.name = this.name().toLowerCase(Locale.ROOT);

connect/api/src/main/java/org/apache/kafka/connect/data/SchemaProjector.java

@@ -54,7 +54,7 @@ public class SchemaProjector {
      * @param record the value to project from source schema to target schema
      * @param target the schema to project the record to
      * @return the projected value with target schema
-     * @throws SchemaProjectorException
+     * @throws SchemaProjectorException if the target schema is not optional and does not have a default value
      */
     public static Object project(Schema source, Object record, Schema target) throws SchemaProjectorException {
         checkMaybeCompatible(source, target);

connect/api/src/main/java/org/apache/kafka/connect/data/Values.java

@@ -100,7 +100,7 @@ public class Values {
 
     private static final Pattern TWO_BACKSLASHES = Pattern.compile("\\\\");
 
-    private static final Pattern DOUBLEQOUTE = Pattern.compile("\"");
+    private static final Pattern DOUBLE_QUOTE = Pattern.compile("\"");
 
     /**
      * Convert the specified value to an {@link Type#BOOLEAN} value. The supplied schema is required if the value is a logical
@@ -584,8 +584,7 @@ public class Values {
                 break;
             case STRUCT:
                 if (value instanceof Struct) {
-                    Struct struct = (Struct) value;
+                    return value;
-                    return struct;
                 }
         }
         throw new DataException("Unable to convert " + value + " (" + value.getClass() + ") to " + toSchema);
@@ -732,7 +731,7 @@ public class Values {
 
     protected static String escape(String value) {
         String replace1 = TWO_BACKSLASHES.matcher(value).replaceAll("\\\\\\\\");
-        return DOUBLEQOUTE.matcher(replace1).replaceAll("\\\\\"");
+        return DOUBLE_QUOTE.matcher(replace1).replaceAll("\\\\\"");
     }
 
     public static DateFormat dateFormatFor(java.util.Date value) {

connect/api/src/main/java/org/apache/kafka/connect/rest/ConnectRestExtension.java

@@ -48,7 +48,7 @@ public interface ConnectRestExtension extends Configurable, Versioned, Closeable
     /**
      * ConnectRestExtension implementations can register custom JAX-RS resources via the {@link #register(ConnectRestExtensionContext)}
      * method. The Connect framework will invoke this method after registering the default Connect resources. If the implementations attempt
-     * to re-register any of the Connect resources, it will be be ignored and will be logged.
+     * to re-register any of the Connect resources, it will be ignored and will be logged.
      *
      * @param restPluginContext The context provides access to JAX-RS {@link javax.ws.rs.core.Configurable} and {@link
      *                          ConnectClusterState}.The custom JAX-RS resources can be registered via the {@link

connect/api/src/main/java/org/apache/kafka/connect/source/ExactlyOnceSupport.java

@@ -27,5 +27,5 @@ public enum ExactlyOnceSupport {
     /**
      * Signals that a connector does not support exactly-once delivery.
      */
-    UNSUPPORTED;
+    UNSUPPORTED
 }

connect/api/src/main/java/org/apache/kafka/connect/storage/ConverterType.java

@@ -52,7 +52,7 @@ public enum ConverterType {
         return NAME_TO_TYPE.get(name.toLowerCase(Locale.getDefault()));
     }
 
-    private String name;
+    private final String name;
 
     ConverterType() {
         this.name = this.name().toLowerCase(Locale.ROOT);

connect/api/src/test/java/org/apache/kafka/connect/source/SourceRecordTest.java

@@ -62,7 +62,7 @@ public class SourceRecordTest {
     }
 
     @Test
-    public void shouldCreateSinkRecordWithEmtpyHeaders() {
+    public void shouldCreateSinkRecordWithEmptyHeaders() {
         assertEquals(SOURCE_PARTITION, record.sourcePartition());
         assertEquals(SOURCE_OFFSET, record.sourceOffset());
         assertEquals(TOPIC_NAME, record.topic());

connect/api/src/test/java/org/apache/kafka/connect/storage/StringConverterTest.java

@@ -32,7 +32,7 @@ public class StringConverterTest {
     private static final String TOPIC = "topic";
     private static final String SAMPLE_STRING = "a string";
 
-    private StringConverter converter = new StringConverter();
+    private final StringConverter converter = new StringConverter();
 
     @Test
     public void testStringToBytes() {

connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java

@@ -284,7 +284,7 @@ public abstract class Cast<R extends ConnectRecord<R>> implements Transformation
                 case STRING:
                     return castToString(value);
                 default:
-                    throw new DataException(targetType.toString() + " is not supported in the Cast transformation.");
+                    throw new DataException(targetType + " is not supported in the Cast transformation.");
             }
         } catch (NumberFormatException e) {
             throw new DataException("Value (" + value.toString() + ") was out of range for requested data type", e);

connect/transforms/src/test/java/org/apache/kafka/connect/transforms/DropHeadersTest.java

@@ -32,7 +32,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
 
 public class DropHeadersTest {
 
-    private DropHeaders<SourceRecord> xform = new DropHeaders<>();
+    private final DropHeaders<SourceRecord> xform = new DropHeaders<>();
 
     private Map<String, ?> config(String... headers) {
         Map<String, Object> result = new HashMap<>();
@@ -109,9 +109,8 @@ public class DropHeadersTest {
         Object value = "value";
         Long timestamp = 0L;
 
-        SourceRecord record = new SourceRecord(sourcePartition, sourceOffset, topic, partition,
+        return new SourceRecord(sourcePartition, sourceOffset, topic, partition,
                 keySchema, key, valueSchema, value, timestamp, headers);
-        return record;
     }
 }
 

connect/transforms/src/test/java/org/apache/kafka/connect/transforms/InsertFieldTest.java

@@ -36,8 +36,8 @@ import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
 public class InsertFieldTest {
-    private InsertField<SourceRecord> xformKey = new InsertField.Key<>();
+    private final InsertField<SourceRecord> xformKey = new InsertField.Key<>();
-    private InsertField<SourceRecord> xformValue = new InsertField.Value<>();
+    private final InsertField<SourceRecord> xformValue = new InsertField.Value<>();
 
     @AfterEach
     public void teardown() {

connect/transforms/src/test/java/org/apache/kafka/connect/transforms/InsertHeaderTest.java

@@ -32,7 +32,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
 
 public class InsertHeaderTest {
 
-    private InsertHeader<SourceRecord> xform = new InsertHeader<>();
+    private final InsertHeader<SourceRecord> xform = new InsertHeader<>();
 
     private Map<String, ?> config(String header, String valueLiteral) {
         Map<String, String> result = new HashMap<>();
@@ -113,9 +113,8 @@ public class InsertHeaderTest {
         Object value = "value";
         Long timestamp = 0L;
 
-        SourceRecord record = new SourceRecord(sourcePartition, sourceOffset, topic, partition,
+        return new SourceRecord(sourcePartition, sourceOffset, topic, partition,
                 keySchema, key, valueSchema, value, timestamp, headers);
-        return record;
     }
 }
 

connect/transforms/src/test/java/org/apache/kafka/connect/transforms/ReplaceFieldTest.java

@@ -30,7 +30,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNull;
 
 public class ReplaceFieldTest {
-    private ReplaceField<SinkRecord> xform = new ReplaceField.Value<>();
+    private final ReplaceField<SinkRecord> xform = new ReplaceField.Value<>();
 
     @AfterEach
     public void teardown() {