Skip to content

MINOR: Permit tombstone offsets with any partition for file source connector #14233

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
C0urante wants to merge 1 commit into from
+23 −5
Closed

MINOR: Permit tombstone offsets with any partition for file source connector #14233

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 8 additions & 5 deletions connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -117,6 +117,14 @@ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String,
// This connector makes use of a single source partition at a time which represents the file that it is configured to read from.
// However, there could also be source partitions from previous configurations of the connector.
for (Map.Entry<Map<String, ?>, Map<String, ?>> partitionOffset : offsets.entrySet()) {
Map<String, ?> offset = partitionOffset.getValue();
// null offsets are allowed and represent a deletion of offsets for a partition
// allow tombstones for anything; if there's garbage in the offsets for the connector, we don't
// want to prevent users from being able to clean it up using the REST API
if (offset == null) {
continue;
}

Map<String, ?> partition = partitionOffset.getKey();
if (partition == null) {
throw new ConnectException("Partition objects cannot be null");
Expand All @@ -126,11 +134,6 @@ public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String,
throw new ConnectException("Partition objects should contain the key '" + FILENAME_FIELD + "'");
}

Map<String, ?> offset = partitionOffset.getValue();
// null offsets are allowed and represent a deletion of offsets for a partition
if (offset == null) {
continue;
}

if (!offset.containsKey(POSITION_FIELD)) {
throw new ConnectException("Offset objects should either be null or contain the key '" + POSITION_FIELD + "'");
Expand Down
15 changes: 15 additions & 0 deletions connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceConnectorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -213,6 +213,21 @@ public void testAlterOffsetsOffsetPositionValues() {
assertThrows(ConnectException.class, () -> alterOffsets.apply(-10L));
assertTrue(() -> alterOffsets.apply(10L));
}
@Test
public void testAlterOffsetsOffsetTombstones() {
Function<Map<String, ?>, Boolean> alterOffsets = partition ->
connector.alterOffsets(sourceProperties, Collections.singletonMap(partition, null));

assertTrue(alterOffsets.apply(null));
assertTrue(alterOffsets.apply(Collections.emptyMap()));
Map<String, Object> partition = new HashMap<>();
partition.put("unused_partition_key", "unused_partition_value");
assertTrue(alterOffsets.apply(partition));
partition.put(FILENAME_FIELD, FILENAME);
assertTrue(alterOffsets.apply(partition));
partition.put("", "");
assertTrue(alterOffsets.apply(partition));
}

@Test
public void testSuccessfulAlterOffsets() {
Expand Down