From ee7c39d8b691eea8f4b5909e0371a9d76bcf34de Mon Sep 17 00:00:00 2001 From: Yash Mayya Date: Mon, 26 Jun 2023 14:57:06 +0530 Subject: [PATCH 1/5] KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector --- .../connect/file/FileStreamSinkConnector.java | 8 +++ .../file/FileStreamSourceConnector.java | 40 +++++++++++ .../file/FileStreamSourceConnectorTest.java | 71 +++++++++++++++++-- 3 files changed, 115 insertions(+), 4 deletions(-) diff --git a/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSinkConnector.java b/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSinkConnector.java index fbe9cfff1dea2..68ee27cb93972 100644 --- a/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSinkConnector.java +++ b/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSinkConnector.java @@ -16,6 +16,7 @@ */ package org.apache.kafka.connect.file; +import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.config.AbstractConfig; import org.apache.kafka.common.config.ConfigDef; import org.apache.kafka.common.config.ConfigDef.Importance; @@ -79,4 +80,11 @@ public void stop() { public ConfigDef config() { return CONFIG_DEF; } + + @Override + public boolean alterOffsets(Map connectorConfig, Map offsets) { + // Nothing to do here since FileStreamSinkConnector does not manage offsets externally nor does it require any + // custom offset validation + return true; + } } diff --git a/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java b/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java index eef6da2b91dac..dc225ecfc5963 100644 --- a/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java +++ b/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java @@ -22,6 +22,7 @@ import org.apache.kafka.common.config.ConfigDef.Type; import org.apache.kafka.common.utils.AppInfoParser; import org.apache.kafka.connect.connector.Task; +import org.apache.kafka.connect.errors.ConnectException; import org.apache.kafka.connect.source.ExactlyOnceSupport; import org.apache.kafka.connect.source.SourceConnector; import org.slf4j.Logger; @@ -31,6 +32,9 @@ import java.util.List; import java.util.Map; +import static org.apache.kafka.connect.file.FileStreamSourceTask.FILENAME_FIELD; +import static org.apache.kafka.connect.file.FileStreamSourceTask.POSITION_FIELD; + /** * Very simple source connector that works with stdin or a file. */ @@ -101,4 +105,40 @@ public ExactlyOnceSupport exactlyOnceSupport(Map props) { : ExactlyOnceSupport.UNSUPPORTED; } + @Override + public boolean alterOffsets(Map connectorConfig, Map, Map> offsets) { + AbstractConfig config = new AbstractConfig(CONFIG_DEF, connectorConfig); + String filename = config.getString(FILE_CONFIG); + if (filename == null || filename.isEmpty()) { + // If the 'file' configuration is unspecified, stdin is used and no offsets are tracked + throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified"); + } + + // An empty offsets map could indicate that the offsets were reset previously or that no offsets have been committed yet (for a reset operation) + // - we don't need any additional validation for this case. + if (offsets.size() == 0) { + return true; + } + + // This connector makes use of a single source partition which represents the file that it is configured to read from + if (offsets.size() > 1) { + throw new ConnectException("The " + FileStreamSourceConnector.class.getSimpleName() + " supports only a single source partition / file"); + } + + Map partition = offsets.keySet().iterator().next(); + if (!partition.containsKey(FILENAME_FIELD)) { + throw new ConnectException("The partition should contain the key '" + FILENAME_FIELD + "'"); + } + if (!filename.equals(partition.get(FILENAME_FIELD))) { + throw new ConnectException("The value for the '" + FILENAME_FIELD + "' key in the partition should match the configured value for the " + + "connector configuration '" + FILE_CONFIG + "'"); + } + + Map offset = offsets.values().iterator().next(); + if (!offset.containsKey(POSITION_FIELD)) { + throw new ConnectException("The offset should contain the key '" + POSITION_FIELD + "'"); + } + // Let the task validate the actual value for the offset position on startup + return true; + } } diff --git a/connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceConnectorTest.java b/connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceConnectorTest.java index d3b0265bc89cd..d0a69c42795a2 100644 --- a/connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceConnectorTest.java +++ b/connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceConnectorTest.java @@ -20,20 +20,24 @@ import org.apache.kafka.common.config.ConfigException; import org.apache.kafka.common.config.ConfigValue; import org.apache.kafka.connect.connector.ConnectorContext; +import org.apache.kafka.connect.errors.ConnectException; import org.apache.kafka.connect.source.ConnectorTransactionBoundaries; import org.apache.kafka.connect.source.ExactlyOnceSupport; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.kafka.connect.file.FileStreamSourceTask.FILENAME_FIELD; +import static org.apache.kafka.connect.file.FileStreamSourceTask.POSITION_FIELD; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.Mockito.mock; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - public class FileStreamSourceConnectorTest { private static final String SINGLE_TOPIC = "test"; @@ -147,4 +151,63 @@ public void testInvalidBatchSize() { sourceProperties.put(FileStreamSourceConnector.TASK_BATCH_SIZE_CONFIG, "abcd"); assertThrows(ConfigException.class, () -> new AbstractConfig(FileStreamSourceConnector.CONFIG_DEF, sourceProperties)); } + + @Test + public void testAlterOffsetsStdin() { + sourceProperties.remove(FileStreamSourceConnector.FILE_CONFIG); + Map, Map> offsets = Collections.singletonMap( + Collections.singletonMap(FILENAME_FIELD, FILENAME), + Collections.singletonMap(POSITION_FIELD, 0) + ); + assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, offsets)); + } + + @Test + public void testAlterOffsetsMultiplePartitions() { + Map, Map> offsets = new HashMap<>(); + offsets.put(Collections.singletonMap(FILENAME_FIELD, FILENAME), Collections.singletonMap(POSITION_FIELD, 0)); + offsets.put(Collections.singletonMap(FILENAME_FIELD, "/someotherfilename"), Collections.singletonMap(POSITION_FIELD, 0)); + assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, offsets)); + } + + @Test + public void testAlterOffsetsIncorrectPartitionKey() { + Map, Map> offsets = Collections.singletonMap( + Collections.singletonMap("invalid_partition_key", FILENAME), + Collections.singletonMap(POSITION_FIELD, 0) + ); + assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, offsets)); + } + + @Test + public void testAlterOffsetsIncorrectPartitionValue() { + Map, Map> offsets = Collections.singletonMap( + Collections.singletonMap(FILENAME_FIELD, "/someotherfilename"), + Collections.singletonMap(POSITION_FIELD, 0) + ); + assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, offsets)); + } + + @Test + public void testAlterOffsetsIncorrectOffsetKey() { + Map, Map> offsets = Collections.singletonMap( + Collections.singletonMap(FILENAME_FIELD, FILENAME), + Collections.singletonMap("invalid_offset_key", 0) + ); + assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, offsets)); + } + + @Test + public void testSuccessfulAlterOffsets() { + Map, Map> offsets = Collections.singletonMap( + Collections.singletonMap(FILENAME_FIELD, FILENAME), + Collections.singletonMap(POSITION_FIELD, 0) + ); + + // Expect no exception to be thrown when a valid offsets map is passed. An empty offsets map is treated as valid + // since it could indicate that the offsets were reset previously or that no offsets have been committed yet + // (for a reset operation) + connector.alterOffsets(sourceProperties, offsets); + connector.alterOffsets(sourceProperties, new HashMap<>()); + } } From 776e3609762db77a87ace6b4d144cf228159a448 Mon Sep 17 00:00:00 2001 From: Yash Mayya Date: Sat, 8 Jul 2023 12:35:10 +0530 Subject: [PATCH 2/5] Relax FileStreamSourceConnector offsets validation criteria to allow for multiple source partitions --- .../file/FileStreamSourceConnector.java | 39 ++++++++----------- .../file/FileStreamSourceConnectorTest.java | 30 +++++++------- 2 files changed, 30 insertions(+), 39 deletions(-) diff --git a/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java b/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java index dc225ecfc5963..56b7ae3c30a5e 100644 --- a/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java +++ b/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java @@ -114,30 +114,25 @@ public boolean alterOffsets(Map connectorConfig, Map, Map> partitionOffset : offsets.entrySet()) { + Map partition = partitionOffset.getKey(); + if (partition == null) { + throw new ConnectException("Partition objects cannot be null"); + } + + if (!partition.containsKey(FILENAME_FIELD)) { + throw new ConnectException("Partition objects should contain the key '" + FILENAME_FIELD + "'"); + } + + Map offset = partitionOffset.getValue(); + // null offsets are allowed and represent a deletion of offsets for a partition + if (offset != null && !offset.containsKey(POSITION_FIELD)) { + throw new ConnectException("Offset objects should either be null or contain the key '" + POSITION_FIELD + "'"); + } } - // This connector makes use of a single source partition which represents the file that it is configured to read from - if (offsets.size() > 1) { - throw new ConnectException("The " + FileStreamSourceConnector.class.getSimpleName() + " supports only a single source partition / file"); - } - - Map partition = offsets.keySet().iterator().next(); - if (!partition.containsKey(FILENAME_FIELD)) { - throw new ConnectException("The partition should contain the key '" + FILENAME_FIELD + "'"); - } - if (!filename.equals(partition.get(FILENAME_FIELD))) { - throw new ConnectException("The value for the '" + FILENAME_FIELD + "' key in the partition should match the configured value for the " + - "connector configuration '" + FILE_CONFIG + "'"); - } - - Map offset = offsets.values().iterator().next(); - if (!offset.containsKey(POSITION_FIELD)) { - throw new ConnectException("The offset should contain the key '" + POSITION_FIELD + "'"); - } // Let the task validate the actual value for the offset position on startup return true; } diff --git a/connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceConnectorTest.java b/connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceConnectorTest.java index d0a69c42795a2..2a5e15543e628 100644 --- a/connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceConnectorTest.java +++ b/connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceConnectorTest.java @@ -162,30 +162,26 @@ public void testAlterOffsetsStdin() { assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, offsets)); } - @Test - public void testAlterOffsetsMultiplePartitions() { - Map, Map> offsets = new HashMap<>(); - offsets.put(Collections.singletonMap(FILENAME_FIELD, FILENAME), Collections.singletonMap(POSITION_FIELD, 0)); - offsets.put(Collections.singletonMap(FILENAME_FIELD, "/someotherfilename"), Collections.singletonMap(POSITION_FIELD, 0)); - assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, offsets)); - } - @Test public void testAlterOffsetsIncorrectPartitionKey() { - Map, Map> offsets = Collections.singletonMap( + assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap( Collections.singletonMap("invalid_partition_key", FILENAME), Collections.singletonMap(POSITION_FIELD, 0) - ); - assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, offsets)); + ))); + + // null partitions are invalid + assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap( + null, + Collections.singletonMap(POSITION_FIELD, 0) + ))); } @Test - public void testAlterOffsetsIncorrectPartitionValue() { - Map, Map> offsets = Collections.singletonMap( - Collections.singletonMap(FILENAME_FIELD, "/someotherfilename"), - Collections.singletonMap(POSITION_FIELD, 0) - ); - assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, offsets)); + public void testAlterOffsetsMultiplePartitions() { + Map, Map> offsets = new HashMap<>(); + offsets.put(Collections.singletonMap(FILENAME_FIELD, FILENAME), Collections.singletonMap(POSITION_FIELD, 0)); + offsets.put(Collections.singletonMap(FILENAME_FIELD, "/someotherfilename"), null); + connector.alterOffsets(sourceProperties, offsets); } @Test From d60ce9eb83a23e3b7bb4e567374d5824fb34fb74 Mon Sep 17 00:00:00 2001 From: Yash Mayya Date: Tue, 11 Jul 2023 15:02:46 +0530 Subject: [PATCH 3/5] Add validation for offset position value; minor test changes --- .../file/FileStreamSourceConnector.java | 21 ++++++-- .../file/FileStreamSourceConnectorTest.java | 54 +++++++++++++++++-- 2 files changed, 67 insertions(+), 8 deletions(-) diff --git a/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java b/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java index 56b7ae3c30a5e..c6ada6ff41c08 100644 --- a/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java +++ b/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java @@ -111,7 +111,8 @@ public boolean alterOffsets(Map connectorConfig, Map connectorConfig, Map offset = partitionOffset.getValue(); // null offsets are allowed and represent a deletion of offsets for a partition - if (offset != null && !offset.containsKey(POSITION_FIELD)) { + if (offset == null) { + return true; + } + + if (!offset.containsKey(POSITION_FIELD)) { throw new ConnectException("Offset objects should either be null or contain the key '" + POSITION_FIELD + "'"); } + + // The 'position' in the offset represents the position in the file's byte stream and should be a non-negative long value + try { + long offsetPosition = Long.parseLong(String.valueOf(offset.get(POSITION_FIELD))); + if (offsetPosition < 0) { + throw new ConnectException("The value for the '" + POSITION_FIELD + "' key in the offset should be a non-negative number"); + } + } catch (NumberFormatException e) { + throw new ConnectException("The value for the '" + POSITION_FIELD + "' key in the offset should be a number"); + } } - // Let the task validate the actual value for the offset position on startup + // Let the task check whether the actual value for the offset position is valid for the configured file on startup return true; } } diff --git a/connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceConnectorTest.java b/connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceConnectorTest.java index 2a5e15543e628..956ff6681ef6f 100644 --- a/connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceConnectorTest.java +++ b/connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceConnectorTest.java @@ -36,6 +36,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.mock; public class FileStreamSourceConnectorTest { @@ -165,7 +166,7 @@ public void testAlterOffsetsStdin() { @Test public void testAlterOffsetsIncorrectPartitionKey() { assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap( - Collections.singletonMap("invalid_partition_key", FILENAME), + Collections.singletonMap("other_partition_key", FILENAME), Collections.singletonMap(POSITION_FIELD, 0) ))); @@ -181,18 +182,61 @@ public void testAlterOffsetsMultiplePartitions() { Map, Map> offsets = new HashMap<>(); offsets.put(Collections.singletonMap(FILENAME_FIELD, FILENAME), Collections.singletonMap(POSITION_FIELD, 0)); offsets.put(Collections.singletonMap(FILENAME_FIELD, "/someotherfilename"), null); - connector.alterOffsets(sourceProperties, offsets); + assertTrue(connector.alterOffsets(sourceProperties, offsets)); } @Test public void testAlterOffsetsIncorrectOffsetKey() { Map, Map> offsets = Collections.singletonMap( Collections.singletonMap(FILENAME_FIELD, FILENAME), - Collections.singletonMap("invalid_offset_key", 0) + Collections.singletonMap("other_offset_key", 0) ); assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, offsets)); } + @Test + public void testAlterOffsetsOffsetPositionValues() { + assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap( + Collections.singletonMap(FILENAME_FIELD, FILENAME), + Collections.singletonMap(POSITION_FIELD, "nan") + ))); + + assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap( + Collections.singletonMap(FILENAME_FIELD, FILENAME), + Collections.singletonMap(POSITION_FIELD, null) + ))); + + assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap( + Collections.singletonMap(FILENAME_FIELD, FILENAME), + Collections.singletonMap(POSITION_FIELD, new Object()) + ))); + + assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap( + Collections.singletonMap(FILENAME_FIELD, FILENAME), + Collections.singletonMap(POSITION_FIELD, 3.14) + ))); + + assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap( + Collections.singletonMap(FILENAME_FIELD, FILENAME), + Collections.singletonMap(POSITION_FIELD, -420) + ))); + + assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap( + Collections.singletonMap(FILENAME_FIELD, FILENAME), + Collections.singletonMap(POSITION_FIELD, "-420") + ))); + + assertTrue(connector.alterOffsets(sourceProperties, Collections.singletonMap( + Collections.singletonMap(FILENAME_FIELD, FILENAME), + Collections.singletonMap(POSITION_FIELD, 10) + ))); + + assertTrue(connector.alterOffsets(sourceProperties, Collections.singletonMap( + Collections.singletonMap(FILENAME_FIELD, FILENAME), + Collections.singletonMap(POSITION_FIELD, "10") + ))); + } + @Test public void testSuccessfulAlterOffsets() { Map, Map> offsets = Collections.singletonMap( @@ -203,7 +247,7 @@ public void testSuccessfulAlterOffsets() { // Expect no exception to be thrown when a valid offsets map is passed. An empty offsets map is treated as valid // since it could indicate that the offsets were reset previously or that no offsets have been committed yet // (for a reset operation) - connector.alterOffsets(sourceProperties, offsets); - connector.alterOffsets(sourceProperties, new HashMap<>()); + assertTrue(connector.alterOffsets(sourceProperties, offsets)); + assertTrue(connector.alterOffsets(sourceProperties, new HashMap<>())); } } From 45f0e539cb9e2afb56bcdaccc7f020a0e9503d9f Mon Sep 17 00:00:00 2001 From: Yash Mayya Date: Tue, 11 Jul 2023 20:00:34 +0530 Subject: [PATCH 4/5] Remove redundant comment; fix early return bug; refactor test --- .../file/FileStreamSourceConnector.java | 3 +- .../file/FileStreamSourceConnectorTest.java | 51 +++++-------------- 2 files changed, 14 insertions(+), 40 deletions(-) diff --git a/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java b/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java index c6ada6ff41c08..c6bc2ffaf01ba 100644 --- a/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java +++ b/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java @@ -110,7 +110,6 @@ public boolean alterOffsets(Map connectorConfig, Map connectorConfig, Map offset = partitionOffset.getValue(); // null offsets are allowed and represent a deletion of offsets for a partition if (offset == null) { - return true; + continue; } if (!offset.containsKey(POSITION_FIELD)) { diff --git a/connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceConnectorTest.java b/connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceConnectorTest.java index 956ff6681ef6f..9702e9bf1e7ee 100644 --- a/connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceConnectorTest.java +++ b/connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceConnectorTest.java @@ -30,6 +30,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.function.Function; import static org.apache.kafka.connect.file.FileStreamSourceTask.FILENAME_FIELD; import static org.apache.kafka.connect.file.FileStreamSourceTask.POSITION_FIELD; @@ -196,45 +197,19 @@ public void testAlterOffsetsIncorrectOffsetKey() { @Test public void testAlterOffsetsOffsetPositionValues() { - assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap( - Collections.singletonMap(FILENAME_FIELD, FILENAME), - Collections.singletonMap(POSITION_FIELD, "nan") - ))); - - assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap( - Collections.singletonMap(FILENAME_FIELD, FILENAME), - Collections.singletonMap(POSITION_FIELD, null) - ))); - - assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap( - Collections.singletonMap(FILENAME_FIELD, FILENAME), - Collections.singletonMap(POSITION_FIELD, new Object()) - ))); - - assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap( - Collections.singletonMap(FILENAME_FIELD, FILENAME), - Collections.singletonMap(POSITION_FIELD, 3.14) - ))); - - assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap( - Collections.singletonMap(FILENAME_FIELD, FILENAME), - Collections.singletonMap(POSITION_FIELD, -420) - ))); - - assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap( + Function alterOffsets = offset -> connector.alterOffsets(sourceProperties, Collections.singletonMap( Collections.singletonMap(FILENAME_FIELD, FILENAME), - Collections.singletonMap(POSITION_FIELD, "-420") - ))); - - assertTrue(connector.alterOffsets(sourceProperties, Collections.singletonMap( - Collections.singletonMap(FILENAME_FIELD, FILENAME), - Collections.singletonMap(POSITION_FIELD, 10) - ))); - - assertTrue(connector.alterOffsets(sourceProperties, Collections.singletonMap( - Collections.singletonMap(FILENAME_FIELD, FILENAME), - Collections.singletonMap(POSITION_FIELD, "10") - ))); + Collections.singletonMap(POSITION_FIELD, offset) + )); + + assertThrows(ConnectException.class, () -> alterOffsets.apply("nan")); + assertThrows(ConnectException.class, () -> alterOffsets.apply(null)); + assertThrows(ConnectException.class, () -> alterOffsets.apply(new Object())); + assertThrows(ConnectException.class, () -> alterOffsets.apply(3.14)); + assertThrows(ConnectException.class, () -> alterOffsets.apply(-420)); + assertThrows(ConnectException.class, () -> alterOffsets.apply("-420")); + assertTrue(() -> alterOffsets.apply(10)); + assertTrue(() -> alterOffsets.apply("10")); } @Test From cd461636bb492d609312e48d97fc7030f048342d Mon Sep 17 00:00:00 2001 From: Yash Mayya Date: Thu, 13 Jul 2023 09:22:41 +0530 Subject: [PATCH 5/5] Refine check for offset position type --- .../file/FileStreamSourceConnector.java | 14 +++++++------- .../file/FileStreamSourceConnectorTest.java | 18 ++++++++++-------- 2 files changed, 17 insertions(+), 15 deletions(-) diff --git a/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java b/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java index c6bc2ffaf01ba..13193f8f50124 100644 --- a/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java +++ b/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java @@ -137,13 +137,13 @@ public boolean alterOffsets(Map connectorConfig, Map, Map> offsets = Collections.singletonMap( Collections.singletonMap(FILENAME_FIELD, FILENAME), - Collections.singletonMap(POSITION_FIELD, 0) + Collections.singletonMap(POSITION_FIELD, 0L) ); assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, offsets)); } @@ -168,20 +168,20 @@ public void testAlterOffsetsStdin() { public void testAlterOffsetsIncorrectPartitionKey() { assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap( Collections.singletonMap("other_partition_key", FILENAME), - Collections.singletonMap(POSITION_FIELD, 0) + Collections.singletonMap(POSITION_FIELD, 0L) ))); // null partitions are invalid assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap( null, - Collections.singletonMap(POSITION_FIELD, 0) + Collections.singletonMap(POSITION_FIELD, 0L) ))); } @Test public void testAlterOffsetsMultiplePartitions() { Map, Map> offsets = new HashMap<>(); - offsets.put(Collections.singletonMap(FILENAME_FIELD, FILENAME), Collections.singletonMap(POSITION_FIELD, 0)); + offsets.put(Collections.singletonMap(FILENAME_FIELD, FILENAME), Collections.singletonMap(POSITION_FIELD, 0L)); offsets.put(Collections.singletonMap(FILENAME_FIELD, "/someotherfilename"), null); assertTrue(connector.alterOffsets(sourceProperties, offsets)); } @@ -190,7 +190,7 @@ public void testAlterOffsetsMultiplePartitions() { public void testAlterOffsetsIncorrectOffsetKey() { Map, Map> offsets = Collections.singletonMap( Collections.singletonMap(FILENAME_FIELD, FILENAME), - Collections.singletonMap("other_offset_key", 0) + Collections.singletonMap("other_offset_key", 0L) ); assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, offsets)); } @@ -208,15 +208,17 @@ public void testAlterOffsetsOffsetPositionValues() { assertThrows(ConnectException.class, () -> alterOffsets.apply(3.14)); assertThrows(ConnectException.class, () -> alterOffsets.apply(-420)); assertThrows(ConnectException.class, () -> alterOffsets.apply("-420")); - assertTrue(() -> alterOffsets.apply(10)); - assertTrue(() -> alterOffsets.apply("10")); + assertThrows(ConnectException.class, () -> alterOffsets.apply(10)); + assertThrows(ConnectException.class, () -> alterOffsets.apply("10")); + assertThrows(ConnectException.class, () -> alterOffsets.apply(-10L)); + assertTrue(() -> alterOffsets.apply(10L)); } @Test public void testSuccessfulAlterOffsets() { Map, Map> offsets = Collections.singletonMap( Collections.singletonMap(FILENAME_FIELD, FILENAME), - Collections.singletonMap(POSITION_FIELD, 0) + Collections.singletonMap(POSITION_FIELD, 0L) ); // Expect no exception to be thrown when a valid offsets map is passed. An empty offsets map is treated as valid