Skip to content

Commit 5b793a9

Browse files
thjarvinthjarvin
committed
Close db query rs and stmt before processing results
1 parent 088bac9 commit 5b793a9

5 files changed

Lines changed: 136 additions & 48 deletions

File tree

‎.github/workflows/test-and-build.yml‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -79,3 +79,22 @@ jobs:
7979
username: ${{ secrets.DOCKER_USERNAME }}
8080
password: ${{ secrets.DOCKER_PASSWORD }}
8181
tags: aks-dev
82+
build-optimize-docker-image:
83+
needs: test
84+
runs-on: ubuntu-latest
85+
# Run only on feat/optimize-query-result-handling branch
86+
if: github.ref == 'refs/heads/feat/optimize-query-result-handling'
87+
steps:
88+
- uses: actions/checkout@v2
89+
- name: Download .jar file
90+
uses: actions/download-artifact@v4
91+
with:
92+
name: transitdata-pubtrans-source.jar
93+
path: target
94+
- name: Build and publish feat/optimize-query-result-handling Docker image
95+
uses: elgohr/Publish-Docker-Github-Action@master
96+
with:
97+
name: hsldevcom/transitdata-pubtrans-source
98+
username: ${{ secrets.DOCKER_USERNAME }}
99+
password: ${{ secrets.DOCKER_PASSWORD }}
100+
tags: feature-branch

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

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -4,15 +4,11 @@
44
import fi.hsl.common.transitdata.TransitdataProperties;
55
import fi.hsl.common.transitdata.TransitdataSchema;
66
import fi.hsl.common.transitdata.proto.PubtransTableProtos;
7-
import org.apache.pulsar.client.api.Producer;
8-
import org.apache.pulsar.client.api.TypedMessageBuilder;
9-
import redis.clients.jedis.Jedis;
107

118
import java.sql.ResultSet;
129
import java.sql.SQLException;
13-
import java.util.LinkedList;
10+
import java.util.Map;
1411
import java.util.Optional;
15-
import java.util.Queue;
1612

1713
public class ArrivalHandler extends PubtransTableHandler {
1814

@@ -35,9 +31,15 @@ protected String getTimetabledDateTimeColumnName() {
3531
protected TransitdataSchema getSchema() {
3632
return schema;
3733
}
34+
35+
@Override
36+
protected Map<String, Long> getTableColumnToIdMap(ResultSet resultSet) throws SQLException {
37+
return Map.of();
38+
}
3839

3940
@Override
40-
protected byte[] createPayload(ResultSet resultSet, PubtransTableProtos.Common common, PubtransTableProtos.DOITripInfo tripInfo) throws SQLException {
41+
protected byte[] createPayload(PubtransTableProtos.Common common, Map<String,
42+
Long> columnToIdMap, PubtransTableProtos.DOITripInfo tripInfo) throws SQLException {
4143
PubtransTableProtos.ROIArrival.Builder arrivalBuilder = PubtransTableProtos.ROIArrival.newBuilder();
4244
arrivalBuilder.setSchemaVersion(arrivalBuilder.getSchemaVersion());
4345
arrivalBuilder.setCommon(common);

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

Lines changed: 27 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -4,21 +4,17 @@
44
import fi.hsl.common.transitdata.TransitdataProperties;
55
import fi.hsl.common.transitdata.TransitdataSchema;
66
import fi.hsl.common.transitdata.proto.PubtransTableProtos;
7-
import org.apache.pulsar.client.api.Producer;
8-
import org.apache.pulsar.client.api.TypedMessageBuilder;
9-
import org.slf4j.Logger;
10-
import org.slf4j.LoggerFactory;
11-
import redis.clients.jedis.Jedis;
127

138
import java.sql.ResultSet;
149
import java.sql.SQLException;
15-
import java.util.LinkedList;
16-
import java.util.Optional;
17-
import java.util.Queue;
10+
import java.util.*;
1811

1912
public class DepartureHandler extends PubtransTableHandler {
2013

2114
static final TransitdataSchema schema;
15+
private static final String COLUMN_HAS_DESTINATION_DISPLAY_ID = "HasDestinationDisplayId";
16+
private static final String COLUMN_HAS_DESTINATION_STOP_AREA_GID = "HasDestinationStopAreaGid";
17+
private static final String COLUMN_HAS_SERVICE_REQUIREMENT_ID = "HasServiceRequirementId";
2218
static {
2319
int defaultVersion = PubtransTableProtos.ROIDeparture.newBuilder().getSchemaVersion();
2420
schema = new TransitdataSchema(TransitdataProperties.ProtobufSchema.PubtransRoiDeparture, Optional.of(defaultVersion));
@@ -37,19 +33,35 @@ protected String getTimetabledDateTimeColumnName() {
3733
protected TransitdataSchema getSchema() {
3834
return schema;
3935
}
36+
37+
@Override
38+
protected Map<String, Long> getTableColumnToIdMap(ResultSet resultSet) throws SQLException {
39+
Map<String, Long> columnToIdMap = new HashMap<>();
40+
if (resultSet.getBytes(COLUMN_HAS_DESTINATION_DISPLAY_ID) != null) {
41+
columnToIdMap.put(COLUMN_HAS_DESTINATION_DISPLAY_ID, resultSet.getLong(COLUMN_HAS_DESTINATION_DISPLAY_ID));
42+
}
43+
if (resultSet.getBytes(COLUMN_HAS_DESTINATION_STOP_AREA_GID) != null) {
44+
columnToIdMap.put(COLUMN_HAS_DESTINATION_STOP_AREA_GID, resultSet.getLong(COLUMN_HAS_DESTINATION_STOP_AREA_GID));
45+
}
46+
if (resultSet.getBytes(COLUMN_HAS_SERVICE_REQUIREMENT_ID) != null) {
47+
columnToIdMap.put(COLUMN_HAS_SERVICE_REQUIREMENT_ID, resultSet.getLong(COLUMN_HAS_SERVICE_REQUIREMENT_ID));
48+
}
49+
return columnToIdMap;
50+
}
4051

4152
@Override
42-
protected byte[] createPayload(ResultSet resultSet, PubtransTableProtos.Common common, PubtransTableProtos.DOITripInfo tripInfo) throws SQLException {
53+
protected byte[] createPayload(PubtransTableProtos.Common common, Map<String, Long> columnToIdMap,
54+
PubtransTableProtos.DOITripInfo tripInfo) throws SQLException {
4355
PubtransTableProtos.ROIDeparture.Builder departureBuilder = PubtransTableProtos.ROIDeparture.newBuilder();
4456
departureBuilder.setSchemaVersion(departureBuilder.getSchemaVersion());
4557
departureBuilder.setCommon(common);
4658
departureBuilder.setTripInfo(tripInfo);
47-
if (resultSet.getBytes("HasDestinationDisplayId") != null)
48-
departureBuilder.setHasDestinationDisplayId(resultSet.getLong("HasDestinationDisplayId"));
49-
if (resultSet.getBytes("HasDestinationStopAreaGid") != null)
50-
departureBuilder.setHasDestinationStopAreaGid(resultSet.getLong("HasDestinationStopAreaGid"));
51-
if (resultSet.getBytes("HasServiceRequirementId") != null)
52-
departureBuilder.setHasServiceRequirementId(resultSet.getLong("HasServiceRequirementId"));
59+
if (columnToIdMap.containsKey(COLUMN_HAS_DESTINATION_DISPLAY_ID))
60+
departureBuilder.setHasDestinationDisplayId(columnToIdMap.get(COLUMN_HAS_DESTINATION_DISPLAY_ID));
61+
if (columnToIdMap.containsKey(COLUMN_HAS_DESTINATION_STOP_AREA_GID))
62+
departureBuilder.setHasDestinationStopAreaGid(columnToIdMap.get(COLUMN_HAS_DESTINATION_STOP_AREA_GID));
63+
if (columnToIdMap.containsKey(COLUMN_HAS_SERVICE_REQUIREMENT_ID))
64+
departureBuilder.setHasServiceRequirementId(columnToIdMap.get(COLUMN_HAS_SERVICE_REQUIREMENT_ID));
5365
PubtransTableProtos.ROIDeparture departure = departureBuilder.build();
5466
return departure.toByteArray();
5567
}

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

Lines changed: 32 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,6 @@
1414
import java.time.format.DateTimeFormatter;
1515
import java.time.temporal.ChronoUnit;
1616
import java.util.Collection;
17-
import java.util.Queue;
1817
import java.util.concurrent.TimeUnit;
1918

2019
public class PubtransConnector {
@@ -109,6 +108,33 @@ static boolean isCacheValid(OffsetDateTime lastCacheUpdate, final int cacheMaxAg
109108
log.info("Current time {}, last update {}} => mins from prev update: {}", now, lastCacheUpdate, minutesSinceUpdate);
110109
return minutesSinceUpdate <= cacheMaxAgeInMins;
111110
}
111+
112+
static void closeQuery(final ResultSet resultSet, final Statement statement) {
113+
if (resultSet == null && statement == null) {
114+
log.warn("ResultSet and Statement are null, nothing to close. {}");
115+
return;
116+
}
117+
try {
118+
if (resultSet.isClosed() && statement.isClosed()) {
119+
log.info("ResultSet and Statement already closed, nothing to close. {}");
120+
return;
121+
}
122+
} catch (SQLException e) {
123+
log.info("Error occured when trying to check if ResultSet and Statement are closed. {}");
124+
}
125+
if (resultSet != null) try {
126+
resultSet.close();
127+
log.info("ResultSet closed. {}");
128+
} catch (Exception e) {
129+
log.error("Failed to close ResultSet", e);
130+
}
131+
if (statement != null) try {
132+
statement.close();
133+
log.info("Statement closed. {}");
134+
} catch (Exception e) {
135+
log.error("Failed to close Statement", e);
136+
}
137+
}
112138

113139
public void queryAndProcessResults() throws SQLException, PulsarClientException {
114140

@@ -122,16 +148,11 @@ public void queryAndProcessResults() throws SQLException, PulsarClientException
122148
statement.setQueryTimeout(queryTimeoutSecs);
123149

124150
resultSet = statement.executeQuery();
125-
126-
produceMessages(handler.handleResultSet(resultSet));
127-
} finally {
128-
if (resultSet != null) try { resultSet.close(); } catch (Exception e) { log.error("Exception while closing result set", e); }
129-
if (statement != null) try { statement.close(); } catch (Exception e) { log.error("Exception while closing statement", e); }
130-
long queryDuration = System.currentTimeMillis() - queryStartTime;
131-
long secondsDuration = queryDuration / 1000;
132-
long minutesDuration = secondsDuration / 60;
133-
long remainingSecondsDuration = secondsDuration % 60;
134-
log.info("Database query executed in {} min {} sec", minutesDuration, remainingSecondsDuration);
151+
152+
produceMessages(handler.handleResultSet(resultSet, statement, queryStartTime));
153+
} catch (PulsarClientException | SQLException e) {
154+
closeQuery(resultSet, statement);
155+
throw e;
135156
}
136157
}
137158

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

Lines changed: 50 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
1010
import org.slf4j.LoggerFactory;
1111
import redis.clients.jedis.Jedis;
1212

13+
import java.sql.PreparedStatement;
1314
import java.sql.ResultSet;
1415
import java.sql.SQLException;
1516
import java.time.LocalDateTime;
@@ -25,6 +26,8 @@ public abstract class PubtransTableHandler {
2526
private Jedis jedis;
2627
private final String timeZone;
2728
private final boolean excludeMetroTrips;
29+
private record QueryResultItem(PubtransTableProtos.Common common, String key, long eventTimestampUtcMs,
30+
Map<String, Long> columnToIdMap) {};
2831

2932
public PubtransTableHandler(PulsarApplicationContext context, TransitdataProperties.ProtobufSchema handlerSchema) {
3033
lastModifiedTimeStamp = (System.currentTimeMillis() - 5000);
@@ -62,35 +65,54 @@ public static Optional<Long> toUtcEpochMs(String localTimestamp, String zoneId)
6265
return Optional.empty();
6366
}
6467
}
68+
69+
abstract protected Map<String, Long> getTableColumnToIdMap(ResultSet resultSet) throws SQLException;
6570

66-
abstract protected byte[] createPayload(ResultSet resultSet, PubtransTableProtos.Common common, PubtransTableProtos.DOITripInfo tripInfo) throws SQLException;
71+
abstract protected byte[] createPayload(
72+
PubtransTableProtos.Common common, Map<String, Long> columnToIdMap,
73+
PubtransTableProtos.DOITripInfo tripInfo) throws SQLException;
6774

6875
abstract protected String getTimetabledDateTimeColumnName();
6976

7077
abstract protected TransitdataSchema getSchema();
7178

72-
public Collection<TypedMessageBuilder<byte[]>> handleResultSet(ResultSet resultSet) throws SQLException {
79+
public Collection<TypedMessageBuilder<byte[]>> handleResultSet(
80+
ResultSet resultSet, PreparedStatement statement, long queryStartTime)
81+
throws SQLException {
7382
List<TypedMessageBuilder<byte[]>> messageBuilderQueue = new ArrayList<>();
7483

7584
long tempTimeStamp = getLastModifiedTimeStamp();
7685

7786
int count = 0;
7887
int metroTripCount = 0;
7988
Set<String> metroRouteIds = new HashSet<>();
89+
List<QueryResultItem> queryResultItems = new ArrayList<>();
90+
long queryDuration = -1L;
91+
long resultHandlerDuration = -1L;
92+
long queryAndResultHandlerDuration = -1L;
8093

8194
while (resultSet.next()) {
8295
count++;
83-
96+
8497
PubtransTableProtos.Common common = parseCommon(resultSet);
8598
final long eventTimestampUtcMs = common.getLastModifiedUtcDateTimeMs();
86-
99+
87100
final long delay = System.currentTimeMillis() - eventTimestampUtcMs;
88101
log.debug("Delay between current time and estimate publish time is {} ms", delay);
89-
102+
90103
final String key = resultSet.getString("IsOnDatedVehicleJourneyId") + resultSet.getString("JourneyPatternSequenceNumber");
91-
final long dvjId = common.getIsOnDatedVehicleJourneyId();
92-
final long scheduledJppId = common.getIsTimetabledAtJourneyPatternPointGid();
93-
final long targetedJppId = common.getIsTargetedAtJourneyPatternPointGid();
104+
final Map<String, Long> columnToIdMap = getTableColumnToIdMap(resultSet);
105+
queryResultItems.add(new QueryResultItem(common, key, eventTimestampUtcMs, columnToIdMap));
106+
}
107+
108+
PubtransConnector.closeQuery(resultSet, statement);
109+
queryDuration = System.currentTimeMillis() - queryStartTime;
110+
111+
for (QueryResultItem queryResultItem : queryResultItems) {
112+
final long resultHandlerStartTime = System.currentTimeMillis();
113+
final long dvjId = queryResultItem.common.getIsOnDatedVehicleJourneyId();
114+
final long scheduledJppId = queryResultItem.common.getIsTimetabledAtJourneyPatternPointGid();
115+
final long targetedJppId = queryResultItem.common.getIsTargetedAtJourneyPatternPointGid();
94116

95117
Optional<PubtransTableProtos.DOITripInfo> maybeTripInfo = getTripInfo(dvjId, scheduledJppId, targetedJppId);
96118
if (maybeTripInfo.isEmpty()) {
@@ -102,21 +124,26 @@ public Collection<TypedMessageBuilder<byte[]>> handleResultSet(ResultSet resultS
102124
metroTripCount++;
103125
metroRouteIds.add(tripInfo.getRouteId());
104126
} else {
105-
final byte[] data = createPayload(resultSet, common, tripInfo);
106-
TypedMessageBuilder<byte[]> msgBuilder = createMessage(key, eventTimestampUtcMs, dvjId, data, getSchema());
127+
final byte[] data = createPayload(queryResultItem.common, queryResultItem.columnToIdMap, tripInfo);
128+
TypedMessageBuilder<byte[]> msgBuilder = createMessage(queryResultItem.key,
129+
queryResultItem.eventTimestampUtcMs, dvjId, data, getSchema());
107130
messageBuilderQueue.add(msgBuilder);
108131
}
109132
}
110133

111134
//Update latest ts for next round
112-
if (eventTimestampUtcMs > tempTimeStamp) {
113-
tempTimeStamp = eventTimestampUtcMs;
135+
if (queryResultItem.eventTimestampUtcMs > tempTimeStamp) {
136+
tempTimeStamp = queryResultItem.eventTimestampUtcMs;
114137
}
138+
resultHandlerDuration = System.currentTimeMillis() - resultHandlerStartTime;
139+
queryAndResultHandlerDuration = System.currentTimeMillis() - queryStartTime;
115140
}
116-
117-
log.info("{} rows processed from the result set. {} rows skipped with metro trips (route ids: {})",
118-
count, metroTripCount, metroRouteIds);
119-
141+
142+
log.info("{} rows processed from the result set. {} rows skipped with metro trips (route ids: {}). "
143+
+ "Operation took {} (db query took {}, handling results took {})",
144+
count, metroTripCount, metroRouteIds,
145+
getMinSec(queryAndResultHandlerDuration), getMinSec(queryDuration), getMinSec(resultHandlerDuration));
146+
120147
setLastModifiedTimeStamp(tempTimeStamp);
121148

122149
return messageBuilderQueue;
@@ -155,6 +182,13 @@ protected PubtransTableProtos.Common parseCommon(ResultSet resultSet) throws SQL
155182
commonBuilder.setLastModifiedUtcDateTimeMs(eventTimestampUtcMs);
156183
return commonBuilder.build();
157184
}
185+
186+
private String getMinSec(long durationMs) {
187+
long seconds = durationMs / 1000;
188+
long minutes = seconds / 60;
189+
long remainingSeconds = seconds % 60;
190+
return String.format("%d min %d sec", minutes, remainingSeconds);
191+
}
158192

159193
private Optional<String> getStopId(long jppId) {
160194
synchronized (jedis) {

0 commit comments

Comments
 (0)