Skip to content

Commit 9ce7ad1

Browse files
chihaiaalexchihaiaalex
authored
Merge pull request #62 from HSLdevcom/feature/61366-redis-cluster
61366 use new transitdata common, connect to redis cluster
2 parents b87dcaa + c14ba05 commit 9ce7ad1

5 files changed

Lines changed: 64 additions & 58 deletions

File tree

‎pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@
1616
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
1717
<maven.compiler.source>24</maven.compiler.source>
1818
<maven.compiler.target>24</maven.compiler.target>
19-
<common.version>2.0.3</common.version>
19+
<common.version>2.0.6</common.version>
2020
</properties>
2121

2222
<dependencies>

‎src/main/java/fi/hsl/transitdata/pulsarpubtransconnect/PubtransConnector.java‎

Lines changed: 20 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -2,13 +2,19 @@
22

33
import com.typesafe.config.Config;
44
import fi.hsl.common.pulsar.PulsarApplicationContext;
5+
import fi.hsl.common.redis.RedisStore;
56
import fi.hsl.common.transitdata.TransitdataProperties;
6-
import org.apache.pulsar.client.api.*;
7+
import org.apache.pulsar.client.api.Producer;
8+
import org.apache.pulsar.client.api.PulsarClientException;
9+
import org.apache.pulsar.client.api.TypedMessageBuilder;
710
import org.slf4j.Logger;
811
import org.slf4j.LoggerFactory;
9-
import redis.clients.jedis.Jedis;
1012

11-
import java.sql.*;
13+
import java.sql.Connection;
14+
import java.sql.PreparedStatement;
15+
import java.sql.ResultSet;
16+
import java.sql.SQLException;
17+
import java.sql.Statement;
1218
import java.time.Duration;
1319
import java.time.OffsetDateTime;
1420
import java.time.format.DateTimeFormatter;
@@ -29,7 +35,7 @@ public class PubtransConnector {
2935
private int queryTimeoutSecs;
3036

3137
private PubtransTableHandler handler;
32-
private Jedis jedis;
38+
private RedisStore redisStore;
3339
private Producer<byte[]> producer;
3440

3541
private PubtransConnector() {}
@@ -40,7 +46,7 @@ public static PubtransConnector newInstance(Connection connection,
4046
PubtransConnector connector = new PubtransConnector();
4147

4248
connector.connection = connection;
43-
connector.jedis = context.getJedis();
49+
connector.redisStore = context.getRedisStore();
4450
connector.producer = context.getSingleProducer();
4551

4652
Config config = context.getConfig();
@@ -84,17 +90,15 @@ private static String queryString(Config config) {
8490
public boolean checkPrecondition() {
8591
if (!enableCacheCheck)
8692
return true;
87-
synchronized (jedis) {
88-
String lastUpdate = jedis.get(TransitdataProperties.KEY_LAST_CACHE_UPDATE_TIMESTAMP);
89-
log.info("Cache last known update: {}", lastUpdate);
90-
if (lastUpdate != null) {
91-
OffsetDateTime dt = OffsetDateTime.parse(lastUpdate, DateTimeFormatter.ISO_DATE_TIME);
92-
return isCacheValid(dt, cacheMaxAgeInMins);
93-
}
94-
else {
95-
log.error("Could not find last cache update timestamp from redis");
96-
return false;
97-
}
93+
94+
var lastUpdate = redisStore.getValue(TransitdataProperties.KEY_LAST_CACHE_UPDATE_TIMESTAMP);
95+
log.info("Cache last known update: {}", lastUpdate);
96+
if (lastUpdate.isPresent()) {
97+
OffsetDateTime dt = OffsetDateTime.parse(lastUpdate.get(), DateTimeFormatter.ISO_DATE_TIME);
98+
return isCacheValid(dt, cacheMaxAgeInMins);
99+
} else {
100+
log.error("Could not find last cache update timestamp from redis");
101+
return false;
98102
}
99103
}
100104

‎src/main/java/fi/hsl/transitdata/pulsarpubtransconnect/PubtransTableHandler.java‎

Lines changed: 35 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -1,37 +1,50 @@
11
package fi.hsl.transitdata.pulsarpubtransconnect;
22

33
import fi.hsl.common.pulsar.PulsarApplicationContext;
4+
import fi.hsl.common.redis.RedisStore;
45
import fi.hsl.common.transitdata.TransitdataProperties;
56
import fi.hsl.common.transitdata.TransitdataSchema;
67
import fi.hsl.common.transitdata.proto.PubtransTableProtos;
78
import org.apache.pulsar.client.api.Producer;
89
import org.apache.pulsar.client.api.TypedMessageBuilder;
910
import org.slf4j.Logger;
1011
import org.slf4j.LoggerFactory;
11-
import redis.clients.jedis.Jedis;
1212

1313
import java.sql.PreparedStatement;
1414
import java.sql.ResultSet;
1515
import java.sql.SQLException;
1616
import java.time.LocalDateTime;
1717
import java.time.ZoneId;
18-
import java.util.*;
18+
import java.util.ArrayList;
19+
import java.util.Collection;
20+
import java.util.HashSet;
21+
import java.util.List;
22+
import java.util.Map;
23+
import java.util.Optional;
24+
import java.util.Set;
25+
26+
import static fi.hsl.common.transitdata.TransitdataProperties.REDIS_PREFIX_DVJ;
27+
import static fi.hsl.common.transitdata.TransitdataProperties.REDIS_PREFIX_JPP;
1928

2029
public abstract class PubtransTableHandler {
2130
static final Logger log = LoggerFactory.getLogger(PubtransTableHandler.class);
2231

2332
private long lastModifiedTimeStamp;
2433
Producer<byte[]> producer;
2534
final TransitdataProperties.ProtobufSchema schema;
26-
private Jedis jedis;
35+
private RedisStore redisStore;
2736
private final String timeZone;
2837
private final boolean excludeMetroTrips;
38+
2939
private record QueryResultItem(PubtransTableProtos.Common common, String key, long eventTimestampUtcMs,
30-
Map<String, Long> columnToIdMap) {};
40+
Map<String, Long> columnToIdMap) {
41+
}
42+
43+
;
3144

3245
public PubtransTableHandler(PulsarApplicationContext context, TransitdataProperties.ProtobufSchema handlerSchema) {
3346
lastModifiedTimeStamp = (System.currentTimeMillis() - 5000);
34-
jedis = context.getJedis();
47+
redisStore = context.getRedisStore();
3548
producer = context.getSingleProducer();
3649
timeZone = context.getConfig().getString("pubtrans.timezone");
3750
excludeMetroTrips = context.getConfig().getBoolean("application.excludeMetroTrips");
@@ -59,17 +72,16 @@ public static Optional<Long> toUtcEpochMs(String localTimestamp, String zoneId)
5972
ZoneId zone = ZoneId.of(zoneId);
6073
long epochMs = dt.atZone(zone).toInstant().toEpochMilli();
6174
return Optional.of(epochMs);
62-
}
63-
catch (Exception e) {
75+
} catch (Exception e) {
6476
log.error("Failed to parse datetime from " + localTimestamp, e);
6577
return Optional.empty();
6678
}
6779
}
68-
80+
6981
public static float getSeconds(long durationMs) {
7082
return durationMs / 1000.0f;
7183
}
72-
84+
7385
abstract protected Map<String, Long> getTableColumnToIdMap(ResultSet resultSet) throws SQLException;
7486

7587
abstract protected byte[] createPayload(
@@ -97,22 +109,22 @@ public Collection<TypedMessageBuilder<byte[]>> handleResultSet(
97109

98110
while (resultSet.next()) {
99111
count++;
100-
112+
101113
PubtransTableProtos.Common common = parseCommon(resultSet);
102114
final long eventTimestampUtcMs = common.getLastModifiedUtcDateTimeMs();
103-
115+
104116
final long delay = System.currentTimeMillis() - eventTimestampUtcMs;
105117
log.debug("Delay between current time and estimate publish time is {} ms", delay);
106-
118+
107119
final String key = resultSet.getString("IsOnDatedVehicleJourneyId") + resultSet.getString("JourneyPatternSequenceNumber");
108120
final Map<String, Long> columnToIdMap = getTableColumnToIdMap(resultSet);
109121
queryResultItems.add(new QueryResultItem(common, key, eventTimestampUtcMs, columnToIdMap));
110122
}
111-
123+
112124
PubtransConnector.closeQuery(resultSet, statement);
113125
long queryEndTime = System.currentTimeMillis();
114126
queryDuration = queryEndTime - queryStartTime;
115-
127+
116128
for (QueryResultItem queryResultItem : queryResultItems) {
117129
final long dvjId = queryResultItem.common.getIsOnDatedVehicleJourneyId();
118130
final long scheduledJppId = queryResultItem.common.getIsTimetabledAtJourneyPatternPointGid();
@@ -123,7 +135,7 @@ public Collection<TypedMessageBuilder<byte[]>> handleResultSet(
123135
log.warn("Could not find valid DOITripInfo from Redis for dvjId {}, timetabledJppId {}, targetedJppId {}. Ignoring this update ", dvjId, scheduledJppId, targetedJppId);
124136
} else {
125137
PubtransTableProtos.DOITripInfo tripInfo = maybeTripInfo.get();
126-
138+
127139
if (excludeMetroTrips && tripInfo.getRouteId().startsWith("31M")) {
128140
metroTripCount++;
129141
metroRouteIds.add(tripInfo.getRouteId());
@@ -140,16 +152,16 @@ public Collection<TypedMessageBuilder<byte[]>> handleResultSet(
140152
tempTimeStamp = queryResultItem.eventTimestampUtcMs;
141153
}
142154
}
143-
155+
144156
long endTime = System.currentTimeMillis();
145157
resultHandlerDuration = endTime - queryEndTime;
146158
queryAndResultHandlerDuration = endTime - queryStartTime;
147-
159+
148160
log.info("{} rows processed from the result set. {} rows skipped with metro trips (route ids: {}). "
149161
+ "Operation took {} s (db query took {} s, handling results took {} s)",
150162
count, metroTripCount, metroRouteIds,
151163
getSeconds(queryAndResultHandlerDuration), getSeconds(queryDuration), getSeconds(resultHandlerDuration));
152-
164+
153165
setLastModifiedTimeStamp(tempTimeStamp);
154166

155167
return messageBuilderQueue;
@@ -190,17 +202,11 @@ protected PubtransTableProtos.Common parseCommon(ResultSet resultSet) throws SQL
190202
}
191203

192204
private Optional<String> getStopId(long jppId) {
193-
synchronized (jedis) {
194-
String stopIdKey = TransitdataProperties.REDIS_PREFIX_JPP + jppId;
195-
return Optional.ofNullable(jedis.get(stopIdKey));
196-
}
205+
return redisStore.getValue(REDIS_PREFIX_JPP + jppId);
197206
}
198207

199208
private Optional<Map<String, String>> getTripInfoFields(long dvjId) {
200-
synchronized (jedis) {
201-
String tripInfoKey = TransitdataProperties.REDIS_PREFIX_DVJ + dvjId;
202-
return Optional.ofNullable(jedis.hgetAll(tripInfoKey));
203-
}
209+
return redisStore.getValues(REDIS_PREFIX_DVJ + dvjId);
204210
}
205211

206212
protected Optional<PubtransTableProtos.DOITripInfo> getTripInfo(long dvjId, long scheduledJppId, long targetedJppId) {
@@ -211,7 +217,7 @@ protected Optional<PubtransTableProtos.DOITripInfo> getTripInfo(long dvjId, long
211217

212218
if (maybeScheduledStopId.isPresent() && maybeTripInfoMap.isPresent()) {
213219
PubtransTableProtos.DOITripInfo.Builder builder = PubtransTableProtos.DOITripInfo.newBuilder();
214-
220+
215221
builder.setStopId(maybeScheduledStopId.get());
216222
maybeTargetedStopId.ifPresent(builder::setTargetedStopId);
217223

@@ -227,13 +233,11 @@ protected Optional<PubtransTableProtos.DOITripInfo> getTripInfo(long dvjId, long
227233
});
228234
builder.setDvjId(dvjId);
229235
return Optional.of(builder.build());
230-
}
231-
else {
236+
} else {
232237
log.error("Failed to get data from Redis for dvjId {}, timetabledJppId {}", dvjId, scheduledJppId);
233238
return Optional.empty();
234239
}
235-
}
236-
catch (Exception e) {
240+
} catch (Exception e) {
237241
log.warn("Failed to get Trip Info for dvj-id " + dvjId, e);
238242
return Optional.empty();
239243
}

‎src/main/resources/arrival.conf‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -11,12 +11,11 @@ pulsar {
1111
}
1212
}
1313

14-
redis {
14+
redisCluster {
1515
enabled=true
16-
connTimeOutSecs=2
17-
connTimeOutSecs=${?REDIS_CONN_TIMEOUT_SECS}
18-
customHealthCheckEnabled="true"
19-
customHealthCheckEnabled=${?REDIS_HEALTH_CHECK_ENABLED}
16+
healthCheck=true
17+
masterName=${?REDIS_CLUSTER_MASTER_NAME}
18+
sentinels=${?REDIS_CLUSTER_SENTINELS}
2019
}
2120

2221
pubtrans {

‎src/main/resources/departure.conf‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -11,12 +11,11 @@ pulsar {
1111
}
1212
}
1313

14-
redis {
14+
redisCluster {
1515
enabled=true
16-
connTimeOutSecs=2
17-
connTimeOutSecs=${?REDIS_CONN_TIMEOUT_SECS}
18-
customHealthCheckEnabled="true"
19-
customHealthCheckEnabled=${?REDIS_HEALTH_CHECK_ENABLED}
16+
healthCheck=true
17+
masterName=${?REDIS_CLUSTER_MASTER_NAME}
18+
sentinels=${?REDIS_CLUSTER_SENTINELS}
2019
}
2120

2221
pubtrans {

0 commit comments

Comments
 (0)