Skip to content

Commit d3295a2

Browse files
author
Oleksii Moskalenko
authored
Redis sink flushes only rows that have more recent eventTimestamp (#913)
* redis sink read then write * fix load tests * e2e * fix * specify feature ref * move test up in order * set project default * some docs * reorder e2e tests * reorder e2e tests * reorder e2e tests
1 parent 294faa5 commit d3295a2

9 files changed

Lines changed: 692 additions & 973 deletions

File tree

‎storage/connectors/redis/pom.xml‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -89,7 +89,12 @@
8989
<version>4.12</version>
9090
<scope>test</scope>
9191
</dependency>
92-
92+
<dependency>
93+
<groupId>org.apache.beam</groupId>
94+
<artifactId>beam-sdks-java-extensions-protobuf</artifactId>
95+
<version>${org.apache.beam.version}</version>
96+
<scope>test</scope>
97+
</dependency>
9398
<dependency>
9499
<groupId>org.slf4j</groupId>
95100
<artifactId>slf4j-simple</artifactId>

‎storage/connectors/redis/src/main/java/feast/storage/connectors/redis/writer/BatchDoFnWithRedis.java‎

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,88 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
* Copyright 2018-2020 The Feast Authors
4+
*
5+
* Licensed under the Apache License, Version 2.0 (the "License");
6+
* you may not use this file except in compliance with the License.
7+
* You may obtain a copy of the License at
8+
*
9+
* https://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package feast.storage.connectors.redis.writer;
18+
19+
import feast.storage.common.retry.Retriable;
20+
import io.lettuce.core.RedisException;
21+
import java.util.concurrent.ExecutionException;
22+
import java.util.concurrent.Future;
23+
import java.util.function.Function;
24+
import org.apache.beam.sdk.transforms.DoFn;
25+
import org.slf4j.Logger;
26+
import org.slf4j.LoggerFactory;
27+
28+
/**
29+
* Base class for redis-related DoFns. Assumes that operations will be batched. Prepares redisClient
30+
* on DoFn.Setup stage and close it on DoFn.Teardown stage.
31+
*
32+
* @param <Input>
33+
* @param <Output>
34+
*/
35+
public class BatchDoFnWithRedis<Input, Output> extends DoFn<Input, Output> {
36+
private static final Logger log = LoggerFactory.getLogger(BatchDoFnWithRedis.class);
37+
38+
private final RedisIngestionClient redisIngestionClient;
39+
40+
BatchDoFnWithRedis(RedisIngestionClient redisIngestionClient) {
41+
this.redisIngestionClient = redisIngestionClient;
42+
}
43+
44+
@Setup
45+
public void setup() {
46+
this.redisIngestionClient.setup();
47+
}
48+
49+
@StartBundle
50+
public void startBundle() {
51+
try {
52+
redisIngestionClient.connect();
53+
} catch (RedisException e) {
54+
log.error("Connection to redis cannot be established: %s", e);
55+
}
56+
}
57+
58+
void executeBatch(Function<RedisIngestionClient, Iterable<Future<? extends Object>>> executor)
59+
throws Exception {
60+
this.redisIngestionClient
61+
.getBackOffExecutor()
62+
.execute(
63+
new Retriable() {
64+
@Override
65+
public void execute() throws ExecutionException, InterruptedException {
66+
if (!redisIngestionClient.isConnected()) {
67+
redisIngestionClient.connect();
68+
}
69+
70+
Iterable<Future<?>> futures = executor.apply(redisIngestionClient);
71+
redisIngestionClient.sync(futures);
72+
}
73+
74+
@Override
75+
public Boolean isExceptionRetriable(Exception e) {
76+
return e instanceof RedisException;
77+
}
78+
79+
@Override
80+
public void cleanUpAfterFailure() {}
81+
});
82+
}
83+
84+
@Teardown
85+
public void teardown() {
86+
redisIngestionClient.shutdown();
87+
}
88+
}

‎storage/connectors/redis/src/main/java/feast/storage/connectors/redis/writer/RedisClusterIngestionClient.java‎

Lines changed: 14 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -20,14 +20,15 @@
2020
import feast.proto.core.StoreProto;
2121
import feast.storage.common.retry.BackOffExecutor;
2222
import io.lettuce.core.LettuceFutures;
23-
import io.lettuce.core.RedisFuture;
2423
import io.lettuce.core.RedisURI;
2524
import io.lettuce.core.cluster.RedisClusterClient;
2625
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
2726
import io.lettuce.core.cluster.api.async.RedisAdvancedClusterAsyncCommands;
2827
import io.lettuce.core.codec.ByteArrayCodec;
2928
import java.util.Arrays;
3029
import java.util.List;
30+
import java.util.concurrent.CompletableFuture;
31+
import java.util.concurrent.Future;
3132
import java.util.concurrent.TimeUnit;
3233
import java.util.stream.Collectors;
3334
import org.joda.time.Duration;
@@ -39,7 +40,6 @@ public class RedisClusterIngestionClient implements RedisIngestionClient {
3940
private transient RedisClusterClient clusterClient;
4041
private StatefulRedisClusterConnection<byte[], byte[]> connection;
4142
private RedisAdvancedClusterAsyncCommands<byte[], byte[]> commands;
42-
private List<RedisFuture> futures = Lists.newArrayList();
4343

4444
public RedisClusterIngestionClient(StoreProto.Store.RedisClusterConfig redisClusterConfig) {
4545
this.uriList =
@@ -55,7 +55,6 @@ public RedisClusterIngestionClient(StoreProto.Store.RedisClusterConfig redisClus
5555
redisClusterConfig.getInitialBackoffMs() > 0 ? redisClusterConfig.getInitialBackoffMs() : 1;
5656
this.backOffExecutor =
5757
new BackOffExecutor(redisClusterConfig.getMaxRetries(), Duration.millis(backoffMs));
58-
this.clusterClient = RedisClusterClient.create(uriList);
5958
}
6059

6160
@Override
@@ -78,6 +77,10 @@ public void connect() {
7877
if (!isConnected()) {
7978
this.connection = clusterClient.connect(new ByteArrayCodec());
8079
this.commands = connection.async();
80+
81+
// despite we're using async API client still flushes after each command by default
82+
// which we don't want since we produce all commands in batches
83+
this.commands.setAutoFlushCommands(false);
8184
}
8285
}
8386

@@ -87,46 +90,20 @@ public boolean isConnected() {
8790
}
8891

8992
@Override
90-
public void sync() {
91-
try {
92-
LettuceFutures.awaitAll(60, TimeUnit.SECONDS, futures.toArray(new RedisFuture[0]));
93-
} finally {
94-
futures.clear();
95-
}
96-
}
97-
98-
@Override
99-
public void pexpire(byte[] key, Long expiryMillis) {
100-
futures.add(commands.pexpire(key, expiryMillis));
101-
}
102-
103-
@Override
104-
public void append(byte[] key, byte[] value) {
105-
futures.add(commands.append(key, value));
106-
}
107-
108-
@Override
109-
public void set(byte[] key, byte[] value) {
110-
futures.add(commands.set(key, value));
111-
}
93+
public void sync(Iterable<Future<?>> futures) {
94+
this.connection.flushCommands();
11295

113-
@Override
114-
public void lpush(byte[] key, byte[] value) {
115-
futures.add(commands.lpush(key, value));
116-
}
117-
118-
@Override
119-
public void rpush(byte[] key, byte[] value) {
120-
futures.add(commands.rpush(key, value));
96+
LettuceFutures.awaitAll(
97+
60, TimeUnit.SECONDS, Lists.newArrayList(futures).toArray(new Future[0]));
12198
}
12299

123100
@Override
124-
public void sadd(byte[] key, byte[] value) {
125-
futures.add(commands.sadd(key, value));
101+
public CompletableFuture<String> set(byte[] key, byte[] value) {
102+
return commands.set(key, value).toCompletableFuture();
126103
}
127104

128105
@Override
129-
public void zadd(byte[] key, Long score, byte[] value) {
130-
futures.add(commands.zadd(key, score, value));
106+
public CompletableFuture<byte[]> get(byte[] key) {
107+
return commands.get(key).toCompletableFuture();
131108
}
132109
}

0 commit comments

Comments
 (0)