This is an automated email from the ASF dual-hosted git repository.
oscerd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-kafka-connector.git
The following commit(s) were added to refs/heads/main by this push:
new 47d918264c Fix #1799: FileTransforms fails the record instead of
emitting a null value (#1810)
47d918264c is described below
commit 47d918264cf786432fcf0d8861d58e35b4319c5e
Author: Andrea Cosentino <[email protected]>
AuthorDate: Mon Aug 24 14:56:03 2026 +0200
Fix #1799: FileTransforms fails the record instead of emitting a null value
(#1810)
FileTransforms.apply() carried an Eclipse-generated catch stub: an
IOException
from readFileToString was printed with e.printStackTrace() and then
execution
continued with c still null, so a file that could not be read was silently
turned into a record with a null value and a schema derived from null.
Propagate it as a ConnectException naming the file instead, so Kafka
Connect's
error handling and DLQ apply and the failure is attributed to the file that
caused it.
Also fix the else branch, which dereferenced r.value() to log the unexpected
type and therefore threw NullPointerException for a record whose value is
null,
and reuse the already-read local instead of calling r.value() three times.
Adds the first test to a connector module. The junit-jupiter test
dependency is
placed before the START OF GENERATED CODE marker in the pom so the connector
generator does not overwrite it; surefire is already configured there.
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
---
connectors/camel-file-kafka-connector/pom.xml | 7 ++
.../file/transformers/FileTransforms.java | 15 ++--
.../file/transformers/FileTransformsTest.java | 88 ++++++++++++++++++++++
3 files changed, 103 insertions(+), 7 deletions(-)
diff --git a/connectors/camel-file-kafka-connector/pom.xml
b/connectors/camel-file-kafka-connector/pom.xml
index f593564622..e6b9a7a95e 100644
--- a/connectors/camel-file-kafka-connector/pom.xml
+++ b/connectors/camel-file-kafka-connector/pom.xml
@@ -49,6 +49,13 @@
<artifactId>commons-io</artifactId>
<version>${commons-io-version}</version>
</dependency>
+ <!-- Test scope; version managed by junit-bom via camel-parent. Kept
outside the generated
+ block below so the connector generator does not overwrite it. -->
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <scope>test</scope>
+ </dependency>
<!--START OF GENERATED CODE-->
<dependency>
<groupId>org.apache.camel.kafkaconnector</groupId>
diff --git
a/connectors/camel-file-kafka-connector/src/main/java/org/apache/camel/kafkaconnector/file/transformers/FileTransforms.java
b/connectors/camel-file-kafka-connector/src/main/java/org/apache/camel/kafkaconnector/file/transformers/FileTransforms.java
index 7c8244119b..cf65a78e00 100644
---
a/connectors/camel-file-kafka-connector/src/main/java/org/apache/camel/kafkaconnector/file/transformers/FileTransforms.java
+++
b/connectors/camel-file-kafka-connector/src/main/java/org/apache/camel/kafkaconnector/file/transformers/FileTransforms.java
@@ -26,6 +26,7 @@ import org.apache.camel.kafkaconnector.utils.SchemaHelper;
import org.apache.commons.io.FileUtils;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.connect.connector.ConnectRecord;
+import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.transforms.Transformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,21 +41,21 @@ public class FileTransforms<R extends ConnectRecord<R>>
implements Transformatio
public R apply(R r) {
Object value = r.value();
- if (r.value() instanceof GenericFile) {
+ if (value instanceof GenericFile) {
LOG.debug("Converting record from RemoteFile to text");
- GenericFile<File> message = (GenericFile<File>)r.value();
- String c = null;
+ GenericFile<File> message = (GenericFile<File>)value;
+ File file = message.getFile();
+ String c;
try {
- c = FileUtils.readFileToString(message.getFile(),
StandardCharsets.UTF_8);
+ c = FileUtils.readFileToString(file, StandardCharsets.UTF_8);
} catch (IOException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
+ throw new ConnectException("Failed to read file " +
file.getAbsolutePath(), e);
}
return r.newRecord(r.topic(), r.kafkaPartition(), null, r.key(),
SchemaHelper.buildSchemaBuilderForType(c), c, r.timestamp());
} else {
- LOG.debug("Unexpected message type: {}", r.value().getClass());
+ LOG.debug("Unexpected message type: {}", value == null ? null :
value.getClass());
return r;
}
diff --git
a/connectors/camel-file-kafka-connector/src/test/java/org/apache/camel/kafkaconnector/file/transformers/FileTransformsTest.java
b/connectors/camel-file-kafka-connector/src/test/java/org/apache/camel/kafkaconnector/file/transformers/FileTransformsTest.java
new file mode 100644
index 0000000000..e9f8ba6044
--- /dev/null
+++
b/connectors/camel-file-kafka-connector/src/test/java/org/apache/camel/kafkaconnector/file/transformers/FileTransformsTest.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.camel.kafkaconnector.file.transformers;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+import org.apache.camel.component.file.GenericFile;
+import org.apache.kafka.connect.errors.ConnectException;
+import org.apache.kafka.connect.source.SourceRecord;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class FileTransformsTest {
+
+ @TempDir
+ private Path tempDir;
+
+ private SourceRecord recordOf(Object value) {
+ return new SourceRecord(null, null, "mytopic", 0, null, null, null,
value, null);
+ }
+
+ private SourceRecord recordFor(File file) {
+ GenericFile<File> genericFile = new GenericFile<>();
+ genericFile.setFile(file);
+ return recordOf(genericFile);
+ }
+
+ @Test
+ public void testReadsFileContentAsString() throws IOException {
+ File file = tempDir.resolve("content.txt").toFile();
+ Files.write(file.toPath(), "hello".getBytes(StandardCharsets.UTF_8));
+
+ SourceRecord transformed = new
FileTransforms<SourceRecord>().apply(recordFor(file));
+
+ assertEquals("hello", transformed.value());
+ assertNotNull(transformed.valueSchema());
+ }
+
+ @Test
+ public void testUnreadableFileFailsTheRecordInsteadOfEmittingANullValue() {
+ File missing = tempDir.resolve("missing.txt").toFile();
+
+ FileTransforms<SourceRecord> transform = new FileTransforms<>();
+ SourceRecord record = recordFor(missing);
+
+ ConnectException e = assertThrows(ConnectException.class, () ->
transform.apply(record));
+
+ assertInstanceOf(IOException.class, e.getCause());
+ }
+
+ @Test
+ public void testRecordWithANullValueIsPassedThrough() {
+ SourceRecord record = recordOf(null);
+
+ assertSame(record, new FileTransforms<SourceRecord>().apply(record));
+ }
+
+ @Test
+ public void testRecordOfAnUnexpectedTypeIsPassedThrough() {
+ SourceRecord record = recordOf("not a file");
+
+ assertSame(record, new FileTransforms<SourceRecord>().apply(record));
+ }
+}
