Skip to content
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

Fixes for 8,9 #10

Merged
merged 4 commits into from
Feb 13, 2019
Merged

Fixes for 8,9 #10

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
eae411a
Added more logging when data does not meet criteria. Added check for …
jcustenborder Feb 13, 2019
09a2407
Corrected formatting.
jcustenborder Feb 13, 2019
5fe5c08
Check if assignments are empty.
jcustenborder Feb 13, 2019
7b8e8ad
If an empty list of records are passed to put(), then no offsets will…
jcustenborder Feb 13, 2019
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
98 changes: 59 additions & 39 deletions src/main/java/com/github/jcustenborder/kafka/connect/redis/RedisSinkTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,35 +73,36 @@ public void start(Map<String, String> settings) {
this.session = RedisSessionImpl.create(this.config);

final Set<TopicPartition> assignment = this.context.assignment();
final byte[][] partitionKeys = assignment.stream()
.map(RedisSinkTask::redisOffsetKey)
.map(s -> s.getBytes(Charsets.UTF_8))
.toArray(byte[][]::new);


final RedisFuture<List<KeyValue<byte[], byte[]>>> partitionKeyFuture = this.session.asyncCommands().mget(partitionKeys);
final List<SinkOffsetState> sinkOffsetStates;
try {
final List<KeyValue<byte[], byte[]>> partitionKey = partitionKeyFuture.get(this.config.operationTimeoutMs, TimeUnit.MILLISECONDS);
sinkOffsetStates = partitionKey.stream()
.map(RedisSinkTask::state)
.filter(Objects::nonNull)
.collect(Collectors.toList());
} catch (InterruptedException | ExecutionException | TimeoutException e) {
throw new RetriableException(e);
}
Map<TopicPartition, Long> partitionOffsets = new HashMap<>(assignment.size());
for (SinkOffsetState state : sinkOffsetStates) {
partitionOffsets.put(state.topicPartition(), state.offset());
log.info("Requesting offset {} for {}", state.offset(), state.topicPartition());
}
for (TopicPartition topicPartition : assignment) {
if (!partitionOffsets.containsKey(topicPartition)) {
partitionOffsets.put(topicPartition, 0L);
log.info("Requesting offset {} for {}", 0L, topicPartition);
if (!assignment.isEmpty()) {
final byte[][] partitionKeys = assignment.stream()
.map(RedisSinkTask::redisOffsetKey)
.map(s -> s.getBytes(Charsets.UTF_8))
.toArray(byte[][]::new);

final RedisFuture<List<KeyValue<byte[], byte[]>>> partitionKeyFuture = this.session.asyncCommands().mget(partitionKeys);
final List<SinkOffsetState> sinkOffsetStates;
try {
final List<KeyValue<byte[], byte[]>> partitionKey = partitionKeyFuture.get(this.config.operationTimeoutMs, TimeUnit.MILLISECONDS);
sinkOffsetStates = partitionKey.stream()
.map(RedisSinkTask::state)
.filter(Objects::nonNull)
.collect(Collectors.toList());
} catch (InterruptedException | ExecutionException | TimeoutException e) {
throw new RetriableException(e);
}
Map<TopicPartition, Long> partitionOffsets = new HashMap<>(assignment.size());
for (SinkOffsetState state : sinkOffsetStates) {
partitionOffsets.put(state.topicPartition(), state.offset());
log.info("Requesting offset {} for {}", state.offset(), state.topicPartition());
}
for (TopicPartition topicPartition : assignment) {
if (!partitionOffsets.containsKey(topicPartition)) {
partitionOffsets.put(topicPartition, 0L);
log.info("Requesting offset {} for {}", 0L, topicPartition);
}
}
this.context.offset(partitionOffsets);
}
this.context.offset(partitionOffsets);
}

private byte[] toBytes(String source, Object input) {
Expand Down Expand Up @@ -129,6 +130,14 @@ private byte[] toBytes(String source, Object input) {
return result;
}

static String formatLocation(SinkRecord record) {
return String.format(
"topic = %s partition = %s offset = %s",
record.topic(),
record.kafkaPartition(),
record.kafkaOffset()
);
}

@Override
public void put(Collection<SinkRecord> records) {
Expand All @@ -140,10 +149,19 @@ public void put(Collection<SinkRecord> records) {
TopicPartitionCounter counter = new TopicPartitionCounter();

for (SinkRecord record : records) {
log.trace("put() - Processing record " + formatLocation(record));
if (null == record.key()) {
throw new DataException("The key for the record cannot be null.");
throw new DataException(
"The key for the record cannot be null. " + formatLocation(record)
);
}
final byte[] key = toBytes("key", record.key());
if (null == key || key.length == 0) {
throw new DataException(
"The key cannot be an empty byte array. " + formatLocation(record)
);
}

final byte[] value = toBytes("value", record.value());

SinkOperation.Type currentOperationType;
Expand Down Expand Up @@ -174,18 +192,20 @@ public void put(Collection<SinkRecord> records) {
);

final List<SinkOffsetState> offsetData = counter.offsetStates();
operation = SinkOperation.create(SinkOperation.Type.SET, this.config, offsetData.size());
operations.add(operation);
for (SinkOffsetState e : offsetData) {
final byte[] key = String.format("__kafka.offset.%s.%s", e.topic(), e.partition()).getBytes(Charsets.UTF_8);
final byte[] value;
try {
value = ObjectMapperFactory.INSTANCE.writeValueAsBytes(e);
} catch (JsonProcessingException e1) {
throw new DataException(e1);
if (!offsetData.isEmpty()) {
operation = SinkOperation.create(SinkOperation.Type.SET, this.config, offsetData.size());
operations.add(operation);
for (SinkOffsetState e : offsetData) {
final byte[] key = String.format("__kafka.offset.%s.%s", e.topic(), e.partition()).getBytes(Charsets.UTF_8);
final byte[] value;
try {
value = ObjectMapperFactory.INSTANCE.writeValueAsBytes(e);
} catch (JsonProcessingException e1) {
throw new DataException(e1);
}
operation.add(key, value);
log.trace("put() - Setting offset: {}", e);
}
operation.add(key, value);
log.trace("put() - Setting offset: {}", e);
}

for (SinkOperation op : operations) {
Expand Down
27 changes: 27 additions & 0 deletions src/test/java/com/github/jcustenborder/kafka/connect/redis/RedisSinkTaskIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
import com.github.jcustenborder.docker.junit5.Compose;
import com.github.jcustenborder.docker.junit5.Port;
import com.google.common.base.Charsets;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import io.lettuce.core.KeyValue;
Expand Down Expand Up @@ -61,6 +62,32 @@ public void before() {
this.task = new RedisSinkTask();
}

@Test
public void emptyAssignment(@Port(container = "redis", internalPort = 6379) InetSocketAddress address) throws ExecutionException, InterruptedException {
log.info("address = {}", address);
final String topic = "putWrite";
SinkTaskContext context = mock(SinkTaskContext.class);
when(context.assignment()).thenReturn(ImmutableSet.of());
this.task.initialize(context);
this.task.start(
ImmutableMap.of(RedisSinkConnectorConfig.HOSTS_CONFIG, String.format("%s:%s", address.getHostString(), address.getPort()))
);
}

@Test
public void putEmpty(@Port(container = "redis", internalPort = 6379) InetSocketAddress address) throws ExecutionException, InterruptedException {
log.info("address = {}", address);
final String topic = "putWrite";
SinkTaskContext context = mock(SinkTaskContext.class);
when(context.assignment()).thenReturn(ImmutableSet.of(new TopicPartition(topic, 1)));
this.task.initialize(context);
this.task.start(
ImmutableMap.of(RedisSinkConnectorConfig.HOSTS_CONFIG, String.format("%s:%s", address.getHostString(), address.getPort()))
);

this.task.put(ImmutableList.of());
}

@Test
public void putWrite(@Port(container = "redis", internalPort = 6379) InetSocketAddress address) throws ExecutionException, InterruptedException {
log.info("address = {}", address);
Expand Down