Skip to content

Commit 1b841ef

Browse files
author
Julien Ruaux
committed
deps: Upgraded LettuceMod and Spring Batch Redis
1 parent 9c7cb98 commit 1b841ef

File tree

7 files changed

+115
-72
lines changed

7 files changed

+115
-72
lines changed

‎pom.xml‎

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -17,20 +17,20 @@
1717
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
1818
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
1919
<java.version>8</java.version>
20-
<kafka.version>3.0.0</kafka.version>
20+
<kafka.version>3.2.0</kafka.version>
2121
<connect-utils.version>[0.7.166,0.7.2000)</connect-utils.version>
2222
<asciidoctor.maven.plugin.version>2.1.0</asciidoctor.maven.plugin.version>
2323
<asciidoctorj.version>2.5.1</asciidoctorj.version>
24-
<asciidoctorj.pdf.version>1.6.0</asciidoctorj.pdf.version>
25-
<awaitility.version>4.1.1</awaitility.version>
24+
<asciidoctorj.pdf.version>2.1.4</asciidoctorj.pdf.version>
25+
<awaitility.version>4.2.0</awaitility.version>
2626
<jruby.version>9.2.17.0</jruby.version>
2727
<junit.version>5.8.1</junit.version>
28-
<lettucemod.version>1.8.1</lettucemod.version>
29-
<mockito.version>4.1.0</mockito.version>
30-
<reactor.version>3.4.13</reactor.version>
31-
<spring-batch-redis.version>2.23.9</spring-batch-redis.version>
32-
<spring-batch.version>4.3.4</spring-batch.version>
33-
<slf4j.version>1.7.32</slf4j.version>
28+
<lettucemod.version>2.19.3</lettucemod.version>
29+
<mockito.version>4.6.1</mockito.version>
30+
<reactor.version>3.4.19</reactor.version>
31+
<spring-batch-redis.version>2.30.3</spring-batch-redis.version>
32+
<spring-batch.version>4.3.6</spring-batch.version>
33+
<slf4j.version>1.7.36</slf4j.version>
3434
<testcontainers-redis.version>1.4.6</testcontainers-redis.version>
3535
<testcontainers.version>1.15.3</testcontainers.version>
3636
<component-owner>redis</component-owner>

‎src/main/java/com/redis/kafka/connect/sink/RedisEnterpriseSinkConfig.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -67,7 +67,7 @@ public enum PushDirection {
6767

6868
public static final String WAIT_TIMEOUT_CONFIG = "redis.wait.timeout";
6969
public static final String WAIT_TIMEOUT_DEFAULT = "1000";
70-
public static final String WAIT_TIMEOUT_DOC = "Timeout for WAIT command.";
70+
public static final String WAIT_TIMEOUT_DOC = "Timeout in millis for WAIT command.";
7171

7272
public static final String TYPE_CONFIG = "redis.type";
7373
public static final String TYPE_DEFAULT = DataType.STREAM.name();

‎src/main/java/com/redis/kafka/connect/sink/RedisEnterpriseSinkTask.java‎

Lines changed: 51 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
package com.redis.kafka.connect.sink;
1717

1818
import java.io.IOException;
19+
import java.time.Duration;
1920
import java.util.ArrayList;
2021
import java.util.Collection;
2122
import java.util.Collections;
@@ -29,7 +30,6 @@
2930
import org.apache.kafka.common.TopicPartition;
3031
import org.apache.kafka.common.config.ConfigException;
3132
import org.apache.kafka.connect.data.Field;
32-
import org.apache.kafka.connect.data.Schema;
3333
import org.apache.kafka.connect.data.Struct;
3434
import org.apache.kafka.connect.errors.ConnectException;
3535
import org.apache.kafka.connect.errors.DataException;
@@ -49,20 +49,20 @@
4949
import com.redis.kafka.connect.RedisEnterpriseSinkConnector;
5050
import com.redis.lettucemod.RedisModulesClient;
5151
import com.redis.spring.batch.RedisItemWriter;
52-
import com.redis.spring.batch.RedisItemWriter.OperationItemWriterBuilder;
53-
import com.redis.spring.batch.RedisItemWriter.RedisItemWriterBuilder;
54-
import com.redis.spring.batch.support.RedisOperation;
55-
import com.redis.spring.batch.support.convert.SampleConverter;
56-
import com.redis.spring.batch.support.convert.ScoredValueConverter;
57-
import com.redis.spring.batch.support.operation.Hset;
58-
import com.redis.spring.batch.support.operation.JsonSet;
59-
import com.redis.spring.batch.support.operation.Lpush;
60-
import com.redis.spring.batch.support.operation.Rpush;
61-
import com.redis.spring.batch.support.operation.Sadd;
62-
import com.redis.spring.batch.support.operation.Set;
63-
import com.redis.spring.batch.support.operation.TsAdd;
64-
import com.redis.spring.batch.support.operation.Xadd;
65-
import com.redis.spring.batch.support.operation.Zadd;
52+
import com.redis.spring.batch.RedisItemWriter.OperationBuilder;
53+
import com.redis.spring.batch.RedisItemWriter.WaitForReplication;
54+
import com.redis.spring.batch.convert.SampleConverter;
55+
import com.redis.spring.batch.convert.ScoredValueConverter;
56+
import com.redis.spring.batch.writer.RedisOperation;
57+
import com.redis.spring.batch.writer.operation.Hset;
58+
import com.redis.spring.batch.writer.operation.JsonSet;
59+
import com.redis.spring.batch.writer.operation.Lpush;
60+
import com.redis.spring.batch.writer.operation.Rpush;
61+
import com.redis.spring.batch.writer.operation.Sadd;
62+
import com.redis.spring.batch.writer.operation.Set;
63+
import com.redis.spring.batch.writer.operation.TsAdd;
64+
import com.redis.spring.batch.writer.operation.Xadd;
65+
import com.redis.spring.batch.writer.operation.Zadd;
6666

6767
import io.lettuce.core.KeyValue;
6868
import io.lettuce.core.api.StatefulRedisConnection;
@@ -90,7 +90,7 @@ public void start(final Map<String, String> props) {
9090
client = RedisModulesClient.create(config.getRedisURI());
9191
connection = client.connect();
9292
jsonConverter = new JsonConverter();
93-
jsonConverter.configure(Collections.singletonMap("schemas.enable", "false"), false);
93+
jsonConverter.configure(Collections.singletonMap("schemas.enable", "false"), false);
9494
writer = writer(client).build();
9595
writer.open(new ExecutionContext());
9696
final java.util.Set<TopicPartition> assignment = this.context.assignment();
@@ -124,14 +124,15 @@ private Collection<SinkOffsetState> offsetStates(java.util.Set<TopicPartition> a
124124
return offsetStates;
125125
}
126126

127-
private RedisItemWriterBuilder<byte[], byte[], SinkRecord> writer(RedisModulesClient client) {
128-
RedisItemWriterBuilder<byte[], byte[], SinkRecord> builder = new OperationItemWriterBuilder<>(client,
127+
private RedisItemWriter.Builder<byte[], byte[], SinkRecord> writer(RedisModulesClient client) {
128+
RedisItemWriter.Builder<byte[], byte[], SinkRecord> builder = new OperationBuilder<>(client,
129129
new ByteArrayCodec()).operation(operation());
130130
if (Boolean.TRUE.equals(config.isMultiexec())) {
131131
builder.multiExec();
132132
}
133133
if (config.getWaitReplicas() > 0) {
134-
builder.waitForReplication(config.getWaitReplicas(), config.getWaitTimeout());
134+
builder.waitForReplication(WaitForReplication.builder().replicas(config.getWaitReplicas())
135+
.timeout(Duration.ofMillis(config.getWaitTimeout())).build());
135136
}
136137
return builder;
137138
}
@@ -169,22 +170,26 @@ private RedisOperation<byte[], byte[], SinkRecord> operation() {
169170
}
170171
}
171172

172-
private byte[] value(SinkRecord record) {
173-
return bytes("value", record.value());
173+
private byte[] value(SinkRecord sinkRecord) {
174+
return bytes("value", sinkRecord.value());
174175
}
175176

176-
private byte[] jsonValue(SinkRecord record) {
177-
if (record.value() == null) {
177+
private byte[] jsonValue(SinkRecord sinkRecord) {
178+
Object value = sinkRecord.value();
179+
if (value == null) {
178180
return null;
179181
}
180-
Schema schema = record.valueSchema();
181-
Object value = record.value();
182-
183-
return jsonConverter.fromConnectData(record.topic(), schema, value);
182+
if (value instanceof byte[]) {
183+
return (byte[]) value;
184+
}
185+
if (value instanceof String) {
186+
return ((String) value).getBytes(config.getCharset());
187+
}
188+
return jsonConverter.fromConnectData(sinkRecord.topic(), sinkRecord.valueSchema(), value);
184189
}
185190

186-
private Long longMember(SinkRecord record) {
187-
Object key = record.key();
191+
private Long longMember(SinkRecord sinkRecord) {
192+
Object key = sinkRecord.key();
188193
if (key == null) {
189194
return null;
190195
}
@@ -195,8 +200,8 @@ private Long longMember(SinkRecord record) {
195200
"The key for the record must be a number. Consider using a single message transformation to transform the data before it is written to Redis.");
196201
}
197202

198-
private Double doubleValue(SinkRecord record) {
199-
Object value = record.value();
203+
private Double doubleValue(SinkRecord sinkRecord) {
204+
Object value = sinkRecord.value();
200205
if (value == null) {
201206
return null;
202207
}
@@ -207,25 +212,25 @@ private Double doubleValue(SinkRecord record) {
207212
"The value for the record must be a number. Consider using a single message transformation to transform the data before it is written to Redis.");
208213
}
209214

210-
private boolean isDelete(SinkRecord record) {
211-
return record.value() == null;
215+
private boolean isDelete(SinkRecord sinkRecord) {
216+
return sinkRecord.value() == null;
212217
}
213218

214-
private byte[] key(SinkRecord record) {
219+
private byte[] key(SinkRecord sinkRecord) {
215220
if (config.getKeyspace().isEmpty()) {
216-
return bytes("key", record.key());
221+
return bytes("key", sinkRecord.key());
217222
}
218-
String keyspace = keyspace(record);
219-
String key = keyspace + config.getSeparator() + String.valueOf(record.key());
223+
String keyspace = keyspace(sinkRecord);
224+
String key = keyspace + config.getSeparator() + String.valueOf(sinkRecord.key());
220225
return key.getBytes(config.getCharset());
221226
}
222227

223-
private byte[] member(SinkRecord record) {
224-
return bytes("key", record.key());
228+
private byte[] member(SinkRecord sinkRecord) {
229+
return bytes("key", sinkRecord.key());
225230
}
226231

227-
private String keyspace(SinkRecord record) {
228-
return config.getKeyspace().replace(RedisEnterpriseSinkConfig.TOKEN_TOPIC, record.topic());
232+
private String keyspace(SinkRecord sinkRecord) {
233+
return config.getKeyspace().replace(RedisEnterpriseSinkConfig.TOKEN_TOPIC, sinkRecord.topic());
229234
}
230235

231236
private byte[] bytes(String source, Object input) {
@@ -243,13 +248,13 @@ private byte[] bytes(String source, Object input) {
243248
source));
244249
}
245250

246-
private byte[] collectionKey(SinkRecord record) {
247-
return keyspace(record).getBytes(config.getCharset());
251+
private byte[] collectionKey(SinkRecord sinkRecord) {
252+
return keyspace(sinkRecord).getBytes(config.getCharset());
248253
}
249254

250255
@SuppressWarnings("unchecked")
251-
private Map<byte[], byte[]> map(SinkRecord record) {
252-
Object value = record.value();
256+
private Map<byte[], byte[]> map(SinkRecord sinkRecord) {
257+
Object value = sinkRecord.value();
253258
if (value == null) {
254259
return null;
255260
}
@@ -272,7 +277,7 @@ private Map<byte[], byte[]> map(SinkRecord record) {
272277
}
273278
return body;
274279
}
275-
throw new ConnectException("Unsupported source value type: " + record.valueSchema().type().name());
280+
throw new ConnectException("Unsupported source value type: " + sinkRecord.valueSchema().type().name());
276281
}
277282

278283
@Override

‎src/main/java/com/redis/kafka/connect/source/KeySourceRecordReader.java‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -12,9 +12,10 @@
1212
import org.springframework.batch.item.ExecutionContext;
1313

1414
import com.redis.lettucemod.RedisModulesClient;
15+
import com.redis.spring.batch.DataStructure;
16+
import com.redis.spring.batch.DataStructure.Type;
1517
import com.redis.spring.batch.RedisItemReader;
16-
import com.redis.spring.batch.support.DataStructure;
17-
import com.redis.spring.batch.support.LiveRedisItemReader;
18+
import com.redis.spring.batch.reader.LiveRedisItemReader;
1819

1920
public class KeySourceRecordReader extends AbstractSourceRecordReader<DataStructure<String>> {
2021

@@ -38,7 +39,7 @@ public KeySourceRecordReader(RedisEnterpriseSourceConfig sourceConfig, Duration
3839

3940
@Override
4041
protected void open(RedisModulesClient client) throws Exception {
41-
reader = RedisItemReader.client(client).dataStructure().inMemoryJobs().live().idleTimeout(idleTimeout)
42+
reader = RedisItemReader.client(client).string().dataStructure().live().idleTimeout(idleTimeout)
4243
.keyPatterns(sourceConfig.getKeyPatterns().toArray(new String[0])).build();
4344
reader.open(new ExecutionContext());
4445
}
@@ -61,7 +62,7 @@ protected SourceRecord convert(DataStructure<String> input) {
6162
}
6263

6364
private Schema schema(DataStructure<String> input) {
64-
if (DataStructure.HASH.equals(input.getType())) {
65+
if (Type.HASH.name().equalsIgnoreCase(input.getType())) {
6566
return HASH_VALUE_SCHEMA;
6667
}
6768
return STRING_VALUE_SCHEMA;

‎src/main/java/com/redis/kafka/connect/source/StreamSourceRecordReader.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@
1515

1616
import com.redis.lettucemod.RedisModulesClient;
1717
import com.redis.spring.batch.RedisItemReader;
18-
import com.redis.spring.batch.support.StreamItemReader;
18+
import com.redis.spring.batch.reader.StreamItemReader;
1919

2020
import io.lettuce.core.StreamMessage;
2121

@@ -45,7 +45,7 @@ public StreamSourceRecordReader(RedisEnterpriseSourceConfig sourceConfig, int ta
4545

4646
@Override
4747
protected void open(RedisModulesClient client) {
48-
reader = RedisItemReader.client(client).stream(sourceConfig.getStreamName())
48+
reader = RedisItemReader.client(client).string().stream(sourceConfig.getStreamName())
4949
.offset(sourceConfig.getStreamOffset()).block(Duration.ofMillis(sourceConfig.getStreamBlock()))
5050
.count(sourceConfig.getBatchSize()).consumerGroup(sourceConfig.getStreamConsumerGroup())
5151
.consumer(consumer).build();

‎src/test/integration/java/com/redis/kafka/connect/RedisEnterpriseSinkTaskIT.java‎

Lines changed: 45 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
import java.util.LinkedHashMap;
1515
import java.util.List;
1616
import java.util.Map;
17+
import java.util.Objects;
1718
import java.util.Set;
1819
import java.util.stream.Collectors;
1920

@@ -37,8 +38,8 @@
3738
import com.github.jcustenborder.kafka.connect.utils.SinkRecordHelper;
3839
import com.redis.kafka.connect.sink.RedisEnterpriseSinkConfig;
3940
import com.redis.kafka.connect.sink.RedisEnterpriseSinkTask;
40-
import com.redis.lettucemod.timeseries.RangeOptions;
4141
import com.redis.lettucemod.timeseries.Sample;
42+
import com.redis.lettucemod.timeseries.TimeRange;
4243
import com.redis.testcontainers.RedisModulesContainer;
4344
import com.redis.testcontainers.RedisServer;
4445
import com.redis.testcontainers.junit.jupiter.AbstractTestcontainersRedisTestBase;
@@ -126,8 +127,8 @@ void putHash(RedisTestContext redis) {
126127
}
127128
}
128129

129-
@SuppressWarnings("unused")
130-
private static class Person {
130+
public static class Person {
131+
131132
private long id;
132133
private String name;
133134
private Set<String> hobbies = new HashSet<>();
@@ -165,10 +166,27 @@ public void setAddress(Address address) {
165166
this.address = address;
166167
}
167168

169+
@Override
170+
public int hashCode() {
171+
return Objects.hash(address, hobbies, id, name);
172+
}
173+
174+
@Override
175+
public boolean equals(Object obj) {
176+
if (this == obj)
177+
return true;
178+
if (obj == null)
179+
return false;
180+
if (getClass() != obj.getClass())
181+
return false;
182+
Person other = (Person) obj;
183+
return Objects.equals(address, other.address) && Objects.equals(hobbies, other.hobbies) && id == other.id
184+
&& Objects.equals(name, other.name);
185+
}
186+
168187
}
169188

170-
@SuppressWarnings("unused")
171-
private static class Address {
189+
public static class Address {
172190

173191
private String street;
174192
private String city;
@@ -207,6 +225,24 @@ public void setZip(String zip) {
207225
this.zip = zip;
208226
}
209227

228+
@Override
229+
public int hashCode() {
230+
return Objects.hash(city, state, street, zip);
231+
}
232+
233+
@Override
234+
public boolean equals(Object obj) {
235+
if (this == obj)
236+
return true;
237+
if (obj == null)
238+
return false;
239+
if (getClass() != obj.getClass())
240+
return false;
241+
Address other = (Address) obj;
242+
return Objects.equals(city, other.city) && Objects.equals(state, other.state)
243+
&& Objects.equals(street, other.street) && Objects.equals(zip, other.zip);
244+
}
245+
210246
}
211247

212248
@ParameterizedTest
@@ -250,13 +286,14 @@ void putJSON(RedisTestContext redis) throws JsonProcessingException {
250286
List<SinkRecord> records = new ArrayList<>();
251287
ObjectMapper mapper = new ObjectMapper();
252288
for (Person person : persons) {
289+
String json = mapper.writeValueAsString(person);
253290
records.add(SinkRecordHelper.write(topic, new SchemaAndValue(Schema.STRING_SCHEMA, person.getId()),
254-
new SchemaAndValue(Schema.STRING_SCHEMA, mapper.writeValueAsString(person))));
291+
new SchemaAndValue(Schema.STRING_SCHEMA, json)));
255292
}
256293
put(topic, RedisEnterpriseSinkConfig.DataType.JSON, redis, records);
257294
for (Person person : persons) {
258295
String json = redis.sync().jsonGet(topic + ":" + person.getId());
259-
assertEquals(mapper.writeValueAsString(person), json);
296+
assertEquals(person, mapper.readValue(json, Person.class));
260297
}
261298
}
262299

@@ -276,7 +313,7 @@ void putTimeSeries(RedisTestContext redis) {
276313
new SchemaAndValue(Schema.FLOAT64_SCHEMA, value)));
277314
}
278315
put(topic, RedisEnterpriseSinkConfig.DataType.TIMESERIES, redis, records);
279-
List<Sample> actualSamples = redis.sync().range(topic, RangeOptions.from(0).to(0).build());
316+
List<Sample> actualSamples = redis.sync().range(topic, TimeRange.unbounded());
280317
assertEquals(expectedSamples.size(), actualSamples.size());
281318
for (int index = 0; index < expectedSamples.size(); index++) {
282319
Sample expectedSample = expectedSamples.get(index);

‎src/test/integration/java/com/redis/kafka/connect/RedisEnterpriseSourceTaskIT.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,8 @@
2222
import com.redis.kafka.connect.source.KeySourceRecordReader;
2323
import com.redis.kafka.connect.source.RedisEnterpriseSourceConfig;
2424
import com.redis.kafka.connect.source.RedisEnterpriseSourceTask;
25-
import com.redis.spring.batch.support.DataStructure;
26-
import com.redis.spring.batch.support.LiveRedisItemReader;
25+
import com.redis.spring.batch.DataStructure;
26+
import com.redis.spring.batch.reader.LiveRedisItemReader;
2727
import com.redis.testcontainers.RedisContainer;
2828
import com.redis.testcontainers.RedisServer;
2929
import com.redis.testcontainers.junit.jupiter.AbstractTestcontainersRedisTestBase;

0 commit comments

Comments
 (0)