Skip to content

KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector #13945

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

Merged
C0urante merged 5 commits into from
Jul 18, 2023
Merged

KAFKA-15121: Implement the alterOffsets method in the FileStreamSourceConnector and the FileStreamSinkConnector #13945

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
ee7c39d
KAFKA-15121: Implement the alterOffsets method in the FileStreamSourc…
yashmayya Jun 26, 2023
776e360
Relax FileStreamSourceConnector offsets validation criteria to allow …
yashmayya Jul 8, 2023
d60ce9e
Add validation for offset position value; minor test changes
yashmayya Jul 11, 2023
45f0e53
Remove redundant comment; fix early return bug; refactor test
yashmayya Jul 11, 2023
cd46163
Refine check for offset position type
yashmayya Jul 13, 2023
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
8 changes: 8 additions & 0 deletions connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSinkConnector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -79,4 +80,11 @@ public void stop() {
public ConfigDef config() {
return CONFIG_DEF;
}

@Override
public boolean alterOffsets(Map<String, String> connectorConfig, Map<TopicPartition, Long> offsets) {
// Nothing to do here since FileStreamSinkConnector does not manage offsets externally nor does it require any
// custom offset validation
return true;
}
}
50 changes: 50 additions & 0 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 @@ -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;
Expand All @@ -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.
*/
Expand Down Expand Up @@ -101,4 +105,50 @@ public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
: ExactlyOnceSupport.UNSUPPORTED;
}

@Override
public boolean alterOffsets(Map<String, String> connectorConfig, Map<Map<String, ?>, Map<String, ?>> 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
Comment thread
yashmayya marked this conversation as resolved.
Outdated
throw new ConnectException("Offsets cannot be modified if the '" + FILE_CONFIG + "' configuration is unspecified. " +
"This is because stdin is used for input and offsets are not tracked.");
}
Comment thread
C0urante marked this conversation as resolved.

// 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, ?> 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<String, ?> offset = partitionOffset.getValue();
// null offsets are allowed and represent a deletion of offsets for a partition
if (offset == null) {
return true;
Comment thread
yashmayya marked this conversation as resolved.
Outdated
}

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)));

@C0urante C0urante Jul 11, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Isn't this too permissive? Based on the testAlterOffsetsOffsetPositionValues test case, this would allow values of both "10" (string) and 10 (number) for the offset. But it looks like the source task class would only work with numbers.

@yashmayya yashmayya Jul 11, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, good catch, thanks. I think it might probably be a bit friendlier if we update the task class instead to do similar parsing, WDYT? I'm okay either way, since the most common use case would be copy pasting the output from GET /offsets and modifying it in which case users would end up using a number rather than a string anyway.

@C0urante C0urante Jul 11, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd prefer to leave the task parsing the same; less work on our part, and less risk of a regression in existing parts of the code base.

@yashmayya yashmayya Jul 12, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I found something interesting when experimenting with this. When passing a request body to the PATCH /offsets endpoint like:

{
  "offsets": [
    {
      "partition": {
        "filename": "/path/to/filename"
      },
      "offset": {
        "position": 20
      }
    }
  ]
}

the position 20 is deserialized to an Integer by Jackson (the JSON library we use in the REST layer for Connect) which seems fine because JSON doesn't have separate types for 32 bit and 64 bit numbers. So, the offsets map that is passed to FileStreamSourceConnector::alterOffsets by the runtime also contains 20 as an Integer value. I initially thought that this would cause the FileStreamSourceTask to fail at startup because it uses an instanceof Long check here (and an Integer value is obviously not an instance of Long). However, interestingly, the task did not fail and doing some debugging revealed that after the offsets are serialized and deserialized by the JsonConverter in OffsetsStorageWriter (in Worker::modifySourceConnectorOffsets) and OffsetsStorageReader respectively, the offsets map that is retrieved by the task on startup through its context contains the position 20 as a Long value.

While this particular case is easily handled by simply accepting Integer values as valid in the FileStreamSourceConnector::alterOffsets method, I'm thinking we probably need to make some changes so that the offsets map passed to source connectors in their alterOffsets method is the same as the offsets map that connectors / tasks will retrieve via the OffsetsStorageReader from their context (otherwise, this could lead to some hard to debug issues in other connectors implementing the SourceConnector::alterOffsets method). The easiest way off the top of my head would probably would be to serialize and deserialize the offsets map using the JsonConverter before invoking SourceConnector::alterOffsets. WDYT?

Furthermore, just checking whether the offset position is an instance of Long (Jackson uses a Long if the number doesn't fit in an Integer) or Integer in the FileStreamSourceConnector::alterOffsets method seems sub-optimal because:

  • To someone just reading through the FileStreamSourceConnector and FileStreamSourceTask classes, accepting Integer instances during validation but requiring Long instances in the actual task would look like a bug since the serialization + deserialization aspect isn't transparent.
  • It's an extremely unlikely scenario, but any changes in Jackson's deser logic could break things here - for instance, if smaller numbers are deserialized into Shorts instead of Integers. Parsing using a combination of String::valueOf and Long::parseLong seems a lot more robust.

@C0urante C0urante Jul 12, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, nice catch! I noticed the discrepancy in numeric types while working on KAFKA-15177 but hadn't even considered the possibility of aligning the types across invocations of alterOffsets and OffsetStorageReader::offset/OffsetStorageReader::offsets.

I think re-deserializing the offsets before passing them to alterOffsets is a great idea. Unless the request body is gigantic there shouldn't be serious performance concerns, and it also acts as a nice preflight check to ensure that the offsets can be successfully propagated to the connector's tasks through the offsets topic.

I still don't love permitting string types for the connector's position offset values--it doesn't seem like a great endorsement of our API if we have to implement workarounds in the file connectors, which are the first example of the connector API that many developers see.

@yashmayya yashmayya Jul 12, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since the type alignment issue seemed like a broader one (i.e. not scoped to the file connectors being touched here), I've created a separate ticket and PR for it.

it doesn't seem like a great endorsement of our API if we have to implement workarounds in the file connectors, which are the first example of the connector API that many developers see.

I'd argue that it isn't really a workaround and that the current check itself is bad. If the (de)serialization happened to use Integer for values that fit in a 32 bit signed type (which would be perfectly valid and is exactly what we see currently before the values are passed through the converter), the current check in the FileStreamSourceTask would cause it to bail.

@C0urante C0urante Jul 12, 2023 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmmm... wouldn't that be a pretty serious breaking change if we accidentally switched up how the JSON converter deserializes integer types? Not just for the file source connector, but for plenty of others.

It feels like it might be a better use of our time to make note of this possibility and ensure that we have sufficient unit testing in place to prevent that kind of regression (I suspect we already do but haven't verified this yet).

Of course, because things aren't interesting enough already--it turns out that there's actually two different scenarios in which tasks observe offsets for their connector. The first, which we're all familiar with, is when they query them using an OffsetStorageReader, which in distributed mode reflects the contents of the offsets topic. The second is when SourceTask::commitRecord is invoked, which carries with it the just-ack'd SourceRecord instance originally provided by the task, including the original in-memory source partition and source offset, which may use types that get lost when written to and read back from the offsets topic.

I don't know if this significantly changes the conversation but it seems subtle and counterintuitive enough to bring up so that we can avoid accidentally breaking connector code that relies on this behavior.

@yashmayya yashmayya Jul 13, 2023 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmmm... wouldn't that be a pretty serious breaking change if we accidentally switched up how the JSON converter deserializes integer types? Not just for the file source connector, but for plenty of others.

Okay, that's fair enough, I've changed the check in FileStreamSourceConnector::alterOffsets to mirror the one made in the task at startup for consistency (and avoided making changes in the existing task logic). This does mean that this PR should be merged after #14003 has been merged (assuming that that approach is acceptable).

I don't know if this significantly changes the conversation but it seems subtle and counterintuitive enough to bring up so that we can avoid accidentally breaking connector code that relies on this behavior.

Hm yeah, that's definitely another interesting one to bring up - however, I'd contend that that one kinda makes sense since we're passing the SourceRecord itself - tasks already deal with SourceRecord and their offsets (and associated types) in their regular lifecycle. It would be highly confusing if the SourceRecord that they get in commitRecord doesn't match the one they dispatched to the framework via poll. Of course, ideally, the offsets that they read via OffsetStorageReader should also not have any type mismatches compared to the SourceRecord ones, but I don't think we'd want to (or safely could) change that at this point.

Since the offsets being altered externally would correspond to the ones that the connector / tasks read at startup, I think it makes sense to align the types across invocations to SourceConnector::alterOffsets and offsets queried from an OffsetStorageReader (and an implicit separate alignment between the SourceRecord's offsets types).

@C0urante C0urante Jul 13, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

💯 Sounds good, thanks for thinking this through.

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 check whether the actual value for the offset position is valid for the configured file on startup
return true;
}
}
111 changes: 107 additions & 4 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 @@ -20,20 +20,25 @@
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.junit.jupiter.api.Assertions.assertTrue;
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";
Expand Down Expand Up @@ -147,4 +152,102 @@ 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<String, ?>, Map<String, ?>> offsets = Collections.singletonMap(
Collections.singletonMap(FILENAME_FIELD, FILENAME),
Collections.singletonMap(POSITION_FIELD, 0)
);
assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, offsets));
}

@Test
public void testAlterOffsetsIncorrectPartitionKey() {
assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap(
Collections.singletonMap("other_partition_key", FILENAME),
Collections.singletonMap(POSITION_FIELD, 0)
)));

// null partitions are invalid
assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, Collections.singletonMap(
null,
Collections.singletonMap(POSITION_FIELD, 0)
)));
}

@Test
public void testAlterOffsetsMultiplePartitions() {
Map<Map<String, ?>, Map<String, ?>> offsets = new HashMap<>();
offsets.put(Collections.singletonMap(FILENAME_FIELD, FILENAME), Collections.singletonMap(POSITION_FIELD, 0));
offsets.put(Collections.singletonMap(FILENAME_FIELD, "/someotherfilename"), null);
assertTrue(connector.alterOffsets(sourceProperties, offsets));
}

@Test
public void testAlterOffsetsIncorrectOffsetKey() {
Map<Map<String, ?>, Map<String, ?>> offsets = Collections.singletonMap(
Collections.singletonMap(FILENAME_FIELD, FILENAME),
Collections.singletonMap("other_offset_key", 0)
);
assertThrows(ConnectException.class, () -> connector.alterOffsets(sourceProperties, offsets));
}

@Test
public void testAlterOffsetsOffsetPositionValues() {
Comment thread
yashmayya marked this conversation as resolved.
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<String, ?>, Map<String, ?>> 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)
assertTrue(connector.alterOffsets(sourceProperties, offsets));
assertTrue(connector.alterOffsets(sourceProperties, new HashMap<>()));
}
}