test: Add KeyToValue integration test

Author: RyanSkraba

src/integration-test/java/io/aiven/kafka/connect/transforms/IntegrationTest.java

@@ -23,6 +23,7 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.ExecutionException;
@@ -144,7 +145,7 @@ void setUp() throws ExecutionException, InterruptedException {
         final NewTopic newTopic = new NewTopic(TestSourceConnector.NEW_TOPIC, 1, (short) 1);
         final NewTopic originalTopicForExtractTopicFromValue =
                 new NewTopic(TopicFromValueSchemaConnector.TOPIC, 1, (short) 1);
-        final NewTopic newTopicForExtractTopicFromValue  =
+        final NewTopic newTopicForExtractTopicFromValue =
                 new NewTopic(TopicFromValueSchemaConnector.NAME, 1, (short) 1);
         adminClient.createTopics(Arrays.asList(originalTopic, newTopic, originalTopicForExtractTopicFromValue,
                 newTopicForExtractTopicFromValue)).all().get();
@@ -234,6 +235,40 @@ void testCaseTransform() throws ExecutionException, InterruptedException, IOExce
         );
     }
 
+    @Test
+    @Timeout(10)
+    void testKeyToValue() throws ExecutionException, InterruptedException, IOException {
+
+        final String topicName = TestKeyToValueConnector.TARGET_TOPIC;
+        adminClient.createTopics(List.of(new NewTopic(topicName, 1, (short) 1))).all().get();
+
+        final Map<String, String> connectorConfig = new HashMap<>();
+        connectorConfig.put("name", "test-source-connector");
+        connectorConfig.put("connector.class", TestKeyToValueConnector.class.getName());
+        connectorConfig.put("key.converter", "org.apache.kafka.connect.json.JsonConverter");
+        connectorConfig.put("value.converter", "org.apache.kafka.connect.json.JsonConverter");
+        connectorConfig.put("tasks.max", "1");
+        connectorConfig.put("transforms", "keyToValue");
+        connectorConfig.put("transforms.keyToValue.case", "upper");
+        connectorConfig.put("transforms.keyToValue.key.fields", "a1,a3");
+        connectorConfig.put("transforms.keyToValue.value.fields", "b1");
+        connectorConfig.put("transforms.keyToValue.type", "io.aiven.kafka.connect.transforms.KeyToValue");
+
+        connectRunner.createConnector(connectorConfig);
+
+        waitForCondition(
+            () -> consumer.endOffsets(Collections.singletonList(new TopicPartition(topicName, 0)))
+                    .values().stream().reduce(Long::sum).map(s -> s == TestKeyToValueConnector.MESSAGES_TO_PRODUCE)
+                    .orElse(false), 5000, "Messages appear in target topic"
+        );
+
+        final String payload = "'payload':{'b1':'a1','b2':'b2','b3':'b3','a3':'a3'}".replace('\'', '"');
+        consumer.subscribe(Collections.singletonList(topicName));
+        for (final ConsumerRecord<byte[], byte[]> consumerRecord : consumer.poll(Duration.ofSeconds(1))) {
+            assertThat(consumerRecord.value()).asString().contains(payload);
+        }
+    }
+
     final void checkMessageTransformInTopic(final TopicPartition topicPartition, final long expectedNumberOfMessages)
             throws InterruptedException, IOException {
         waitForCondition(

src/integration-test/java/io/aiven/kafka/connect/transforms/TestKeyToValueConnector.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2025 Aiven Oy
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.aiven.kafka.connect.transforms;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.kafka.connect.connector.Task;
+import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.data.SchemaBuilder;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+import org.apache.kafka.connect.source.SourceTask;
+
+public class TestKeyToValueConnector extends AbstractTestSourceConnector {
+
+    static final long MESSAGES_TO_PRODUCE = 10L;
+
+    static final String TARGET_TOPIC = "key-to-value-target-topic";
+
+    @Override
+    public Class<? extends Task> taskClass() {
+        return TestKeyToValueConnector.TestSourceConnectorTask.class;
+    }
+
+    public static class TestSourceConnectorTask extends SourceTask {
+        private int counter = 0;
+
+        private final Schema keySchema = SchemaBuilder.struct().field("a1", SchemaBuilder.STRING_SCHEMA)
+                .field("a2", SchemaBuilder.STRING_SCHEMA)
+                .field("a3", SchemaBuilder.STRING_SCHEMA).schema();
+        private final Struct key = new Struct(keySchema).put("a1", "a1").put("a2", "a2").put("a3", "a3");
+        private final Schema valueSchema = SchemaBuilder.struct().field("b1", SchemaBuilder.STRING_SCHEMA)
+                .field("b2", SchemaBuilder.STRING_SCHEMA)
+                .field("b3", SchemaBuilder.STRING_SCHEMA).schema();
+        private final Struct value = new Struct(valueSchema).put("b1", "b1").put("b2", "b2").put("b3", "b3");
+
+        @Override
+        public void start(final Map<String, String> props) {
+        }
+
+        @Override
+        public List<SourceRecord> poll() {
+            if (counter >= MESSAGES_TO_PRODUCE) {
+                return null; // indicate pause
+            }
+
+            final Map<String, String> sourcePartition = new HashMap<>();
+            sourcePartition.put("partition", "0");
+            final Map<String, String> sourceOffset = new HashMap<>();
+            sourceOffset.put("offset", Integer.toString(counter));
+
+            counter += 1;
+
+            return Collections.singletonList(
+                    new SourceRecord(sourcePartition, sourceOffset,
+                            TARGET_TOPIC,
+                            keySchema, key,
+                            valueSchema, value)
+            );
+        }
+
+        @Override
+        public void stop() {
+        }
+
+        @Override
+        public String version() {
+            return null;
+        }
+    }
+}