KAFKA-4929: Transformation Key/Value type references should be to class name(), not canonicalName()

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/PluginDiscovery.java

@@ -106,7 +106,7 @@ private static List<Class<? extends Transformation>> transformationPlugins(Refle
         Collections.sort(transformationPlugins, new Comparator<Class<? extends Transformation>>() {
             @Override
             public int compare(Class<? extends Transformation> a, Class<? extends Transformation> b) {
-                return a.getCanonicalName().compareTo(b.getCanonicalName());
+                return a.getName().compareTo(b.getName());
             }
         });
 

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

@@ -31,8 +31,8 @@ public abstract class ExtractField<R extends ConnectRecord<R>> implements Transf
 
     public static final String OVERVIEW_DOC =
             "Extract the specified field from a Struct when schema present, or a Map in the case of schemaless data."
-                    + "<p/>Use the concrete transformation type designed for the record key (<code>" + Key.class.getCanonicalName() + "</code>) "
-                    + "or value (<code>" + Value.class.getCanonicalName() + "</code>).";
+                    + "<p/>Use the concrete transformation type designed for the record key (<code>" + Key.class.getName() + "</code>) "
+                    + "or value (<code>" + Value.class.getName() + "</code>).";
 
     private static final String FIELD_CONFIG = "field";
 

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

@@ -33,8 +33,8 @@ public abstract class HoistField<R extends ConnectRecord<R>> implements Transfor
 
     public static final String OVERVIEW_DOC =
             "Wrap data using the specified field name in a Struct when schema present, or a Map in the case of schemaless data."
-                    + "<p/>Use the concrete transformation type designed for the record key (<code>" + Key.class.getCanonicalName() + "</code>) "
-                    + "or value (<code>" + Value.class.getCanonicalName() + "</code>).";
+                    + "<p/>Use the concrete transformation type designed for the record key (<code>" + Key.class.getName() + "</code>) "
+                    + "or value (<code>" + Value.class.getName() + "</code>).";
 
     private static final String FIELD_CONFIG = "field";
 

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

@@ -42,8 +42,8 @@ public abstract class InsertField<R extends ConnectRecord<R>> implements Transfo
 
     public static final String OVERVIEW_DOC =
             "Insert field(s) using attributes from the record metadata or a configured static value."
-                    + "<p/>Use the concrete transformation type designed for the record key (<code>" + Key.class.getCanonicalName() + "</code>) "
-                    + "or value (<code>" + Value.class.getCanonicalName() + "</code>).";
+                    + "<p/>Use the concrete transformation type designed for the record key (<code>" + Key.class.getName() + "</code>) "
+                    + "or value (<code>" + Value.class.getName() + "</code>).";
 
     private interface ConfigName {
         String TOPIC_FIELD = "topic.field";

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

@@ -42,8 +42,8 @@ public abstract class MaskField<R extends ConnectRecord<R>> implements Transform
 
     public static final String OVERVIEW_DOC =
             "Mask specified fields with a valid null value for the field type (i.e. 0, false, empty string, and so on)."
-                    + "<p/>Use the concrete transformation type designed for the record key (<code>" + Key.class.getCanonicalName() + "</code>) "
-                    + "or value (<code>" + Value.class.getCanonicalName() + "</code>).";
+                    + "<p/>Use the concrete transformation type designed for the record key (<code>" + Key.class.getName() + "</code>) "
+                    + "or value (<code>" + Value.class.getName() + "</code>).";
 
     private static final String FIELDS_CONFIG = "fields";
 

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

@@ -39,7 +39,9 @@
 
 public abstract class ReplaceField<R extends ConnectRecord<R>> implements Transformation<R> {
 
-    public static final String OVERVIEW_DOC = "Filter or rename fields.";
+    public static final String OVERVIEW_DOC = "Filter or rename fields."
+            + "<p/>Use the concrete transformation type designed for the record key (<code>" + Key.class.getName() + "</code>) "
+            + "or value (<code>" + Value.class.getName() + "</code>).";
 
     interface ConfigName {
         String BLACKLIST = "blacklist";

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

@@ -30,8 +30,8 @@
 public abstract class SetSchemaMetadata<R extends ConnectRecord<R>> implements Transformation<R> {
 
     public static final String OVERVIEW_DOC =
-            "Set the schema name, version or both on the record's key (<code>" + Key.class.getCanonicalName() + "</code>)"
-                    + " or value (<code>" + Value.class.getCanonicalName() + "</code>) schema.";
+            "Set the schema name, version or both on the record's key (<code>" + Key.class.getName() + "</code>)"
+                    + " or value (<code>" + Value.class.getName() + "</code>) schema.";
 
     private interface ConfigName {
         String SCHEMA_NAME = "schema.name";