Close db query rs and stmt before processing results

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

@@ -7,11 +7,11 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v2
-      - name: Set up JDK 11
+      - name: Set up JDK 24
         uses: actions/setup-java@v3
         with:
           distribution: 'temurin'
-          java-version: '11'
+          java-version: '24'
           cache: 'maven'
       - name: Build with Maven
         run: mvn --file pom.xml clean install
@@ -79,3 +79,22 @@ jobs:
           username: ${{ secrets.DOCKER_USERNAME }}
           password: ${{ secrets.DOCKER_PASSWORD }}
           tags: aks-dev
+  build-feature-branch-docker-image:
+    needs: test
+    runs-on: ubuntu-latest
+    # Run only on branches starting with feature/
+    if: startsWith(github.ref, 'refs/heads/feature/')
+    steps:
+      - uses: actions/checkout@v2
+      - name: Download .jar file
+        uses: actions/download-artifact@v4
+        with:
+          name: transitdata-pubtrans-source.jar
+          path: target
+      - name: Build and publish feature branch Docker image
+        uses: elgohr/Publish-Docker-Github-Action@master
+        with:
+          name: hsldevcom/transitdata-pubtrans-source
+          username: ${{ secrets.DOCKER_USERNAME }}
+          password: ${{ secrets.DOCKER_PASSWORD }}
+          tags: feature-branch

Dockerfile

@@ -1,4 +1,4 @@
-FROM eclipse-temurin:11-alpine
+FROM eclipse-temurin:24-alpine
 #Install curl for health check
 RUN apk add --no-cache curl
 ADD target/transitdata-pubtrans-source.jar /usr/app/transitdata-pubtrans-source.jar

pom.xml

@@ -14,8 +14,8 @@
 
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <maven.compiler.source>11</maven.compiler.source>
-        <maven.compiler.target>11</maven.compiler.target>
+        <maven.compiler.source>24</maven.compiler.source>
+        <maven.compiler.target>24</maven.compiler.target>
         <common.version>2.0.3</common.version>
     </properties>
 
@@ -103,8 +103,8 @@
                 <artifactId>maven-compiler-plugin</artifactId>
                 <version>3.8.0</version>
                 <configuration>
-                    <source>11</source>
-                    <target>11</target>
+                    <source>24</source>
+                    <target>24</target>
                 </configuration>
             </plugin>
 

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

@@ -4,15 +4,11 @@
 import fi.hsl.common.transitdata.TransitdataProperties;
 import fi.hsl.common.transitdata.TransitdataSchema;
 import fi.hsl.common.transitdata.proto.PubtransTableProtos;
-import org.apache.pulsar.client.api.Producer;
-import org.apache.pulsar.client.api.TypedMessageBuilder;
-import redis.clients.jedis.Jedis;
 
 import java.sql.ResultSet;
 import java.sql.SQLException;
-import java.util.LinkedList;
+import java.util.Map;
 import java.util.Optional;
-import java.util.Queue;
 
 public class ArrivalHandler extends PubtransTableHandler {
 
@@ -35,9 +31,15 @@ protected String getTimetabledDateTimeColumnName() {
     protected TransitdataSchema getSchema() {
         return schema;
     }
+
+    @Override
+    protected Map<String, Long> getTableColumnToIdMap(ResultSet resultSet) throws SQLException {
+        return Map.of();
+    }
 
     @Override
-    protected byte[] createPayload(ResultSet resultSet, PubtransTableProtos.Common common, PubtransTableProtos.DOITripInfo tripInfo) throws SQLException {
+    protected byte[] createPayload(PubtransTableProtos.Common common, Map<String,
+            Long> columnToIdMap, PubtransTableProtos.DOITripInfo tripInfo) throws SQLException {
         PubtransTableProtos.ROIArrival.Builder arrivalBuilder = PubtransTableProtos.ROIArrival.newBuilder();
         arrivalBuilder.setSchemaVersion(arrivalBuilder.getSchemaVersion());
         arrivalBuilder.setCommon(common);

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

@@ -4,21 +4,17 @@
 import fi.hsl.common.transitdata.TransitdataProperties;
 import fi.hsl.common.transitdata.TransitdataSchema;
 import fi.hsl.common.transitdata.proto.PubtransTableProtos;
-import org.apache.pulsar.client.api.Producer;
-import org.apache.pulsar.client.api.TypedMessageBuilder;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import redis.clients.jedis.Jedis;
 
 import java.sql.ResultSet;
 import java.sql.SQLException;
-import java.util.LinkedList;
-import java.util.Optional;
-import java.util.Queue;
+import java.util.*;
 
 public class DepartureHandler extends PubtransTableHandler {
 
     static final TransitdataSchema schema;
+    private static final String COLUMN_HAS_DESTINATION_DISPLAY_ID = "HasDestinationDisplayId";
+    private static final String COLUMN_HAS_DESTINATION_STOP_AREA_GID = "HasDestinationStopAreaGid";
+    private static final String COLUMN_HAS_SERVICE_REQUIREMENT_ID = "HasServiceRequirementId";
     static {
         int defaultVersion = PubtransTableProtos.ROIDeparture.newBuilder().getSchemaVersion();
         schema = new TransitdataSchema(TransitdataProperties.ProtobufSchema.PubtransRoiDeparture, Optional.of(defaultVersion));
@@ -37,19 +33,35 @@ protected String getTimetabledDateTimeColumnName() {
     protected TransitdataSchema getSchema() {
         return schema;
     }
+
+    @Override
+    protected Map<String, Long> getTableColumnToIdMap(ResultSet resultSet) throws SQLException {
+        Map<String, Long> columnToIdMap = new HashMap<>();
+        if (resultSet.getBytes(COLUMN_HAS_DESTINATION_DISPLAY_ID) != null) {
+            columnToIdMap.put(COLUMN_HAS_DESTINATION_DISPLAY_ID, resultSet.getLong(COLUMN_HAS_DESTINATION_DISPLAY_ID));
+        }
+        if (resultSet.getBytes(COLUMN_HAS_DESTINATION_STOP_AREA_GID) != null) {
+            columnToIdMap.put(COLUMN_HAS_DESTINATION_STOP_AREA_GID, resultSet.getLong(COLUMN_HAS_DESTINATION_STOP_AREA_GID));
+        }
+        if (resultSet.getBytes(COLUMN_HAS_SERVICE_REQUIREMENT_ID) != null) {
+            columnToIdMap.put(COLUMN_HAS_SERVICE_REQUIREMENT_ID, resultSet.getLong(COLUMN_HAS_SERVICE_REQUIREMENT_ID));
+        }
+        return columnToIdMap;
+    }
 
     @Override
-    protected byte[] createPayload(ResultSet resultSet, PubtransTableProtos.Common common, PubtransTableProtos.DOITripInfo tripInfo) throws SQLException {
+    protected byte[] createPayload(PubtransTableProtos.Common common, Map<String, Long> columnToIdMap,
+                                   PubtransTableProtos.DOITripInfo tripInfo) throws SQLException {
         PubtransTableProtos.ROIDeparture.Builder departureBuilder = PubtransTableProtos.ROIDeparture.newBuilder();
         departureBuilder.setSchemaVersion(departureBuilder.getSchemaVersion());
         departureBuilder.setCommon(common);
         departureBuilder.setTripInfo(tripInfo);
-        if (resultSet.getBytes("HasDestinationDisplayId") != null)
-            departureBuilder.setHasDestinationDisplayId(resultSet.getLong("HasDestinationDisplayId"));
-        if (resultSet.getBytes("HasDestinationStopAreaGid") != null)
-            departureBuilder.setHasDestinationStopAreaGid(resultSet.getLong("HasDestinationStopAreaGid"));
-        if (resultSet.getBytes("HasServiceRequirementId") != null)
-            departureBuilder.setHasServiceRequirementId(resultSet.getLong("HasServiceRequirementId"));
+        if (columnToIdMap.containsKey(COLUMN_HAS_DESTINATION_DISPLAY_ID))
+            departureBuilder.setHasDestinationDisplayId(columnToIdMap.get(COLUMN_HAS_DESTINATION_DISPLAY_ID));
+        if (columnToIdMap.containsKey(COLUMN_HAS_DESTINATION_STOP_AREA_GID))
+            departureBuilder.setHasDestinationStopAreaGid(columnToIdMap.get(COLUMN_HAS_DESTINATION_STOP_AREA_GID));
+        if (columnToIdMap.containsKey(COLUMN_HAS_SERVICE_REQUIREMENT_ID))
+            departureBuilder.setHasServiceRequirementId(columnToIdMap.get(COLUMN_HAS_SERVICE_REQUIREMENT_ID));
         PubtransTableProtos.ROIDeparture departure = departureBuilder.build();
         return departure.toByteArray();
     }

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

@@ -14,7 +14,6 @@
 import java.time.format.DateTimeFormatter;
 import java.time.temporal.ChronoUnit;
 import java.util.Collection;
-import java.util.Queue;
 import java.util.concurrent.TimeUnit;
 
 public class PubtransConnector {
@@ -109,6 +108,28 @@ static boolean isCacheValid(OffsetDateTime lastCacheUpdate, final int cacheMaxAg
         log.info("Current time {}, last update {}} => mins from prev update: {}", now, lastCacheUpdate, minutesSinceUpdate);
         return minutesSinceUpdate <= cacheMaxAgeInMins;
     }
+
+    static void closeQuery(final ResultSet resultSet, final Statement statement) {
+        if (resultSet != null) {
+            try {
+                resultSet.close();
+                log.debug("ResultSet closed.");
+            } catch (SQLException e) {
+                log.error("Failed to close ResultSet", e);
+            }
+        }
+        if (statement != null) {
+            try {
+                statement.close();
+                log.debug("Statement closed.");
+            } catch (SQLException e) {
+                log.error("Failed to close Statement", e);
+            }
+        }
+        if (resultSet == null && statement == null) {
+            log.warn("ResultSet and Statement are null, nothing to close.");
+        }
+    }
 
     public void queryAndProcessResults() throws SQLException, PulsarClientException {
 
@@ -122,16 +143,11 @@ public void queryAndProcessResults() throws SQLException, PulsarClientException
             statement.setQueryTimeout(queryTimeoutSecs);
 
             resultSet = statement.executeQuery();
-
-            produceMessages(handler.handleResultSet(resultSet));
-        } finally {
-            if (resultSet != null)  try { resultSet.close(); } catch (Exception e) { log.error("Exception while closing result set", e); }
-            if (statement != null)  try { statement.close(); } catch (Exception e) { log.error("Exception while closing statement", e); }
-            long queryDuration = System.currentTimeMillis() - queryStartTime;
-            long secondsDuration = queryDuration / 1000;
-            long minutesDuration = secondsDuration / 60;
-            long remainingSecondsDuration = secondsDuration % 60;
-            log.info("Database query executed in {} min {} sec", minutesDuration, remainingSecondsDuration);
+
+            produceMessages(handler.handleResultSet(resultSet, statement, queryStartTime));
+        } catch (PulsarClientException | SQLException e) {
+            closeQuery(resultSet, statement);
+            throw e;
         }
     }
 

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

@@ -10,6 +10,7 @@
 import org.slf4j.LoggerFactory;
 import redis.clients.jedis.Jedis;
 
+import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.time.LocalDateTime;
@@ -25,6 +26,8 @@ public abstract class PubtransTableHandler {
     private Jedis jedis;
     private final String timeZone;
     private final boolean excludeMetroTrips;
+    private record QueryResultItem(PubtransTableProtos.Common common, String key, long eventTimestampUtcMs,
+                                   Map<String, Long> columnToIdMap) {};
 
     public PubtransTableHandler(PulsarApplicationContext context, TransitdataProperties.ProtobufSchema handlerSchema) {
         lastModifiedTimeStamp = (System.currentTimeMillis() - 5000);
@@ -62,35 +65,66 @@ public static Optional<Long> toUtcEpochMs(String localTimestamp, String zoneId)
             return Optional.empty();
         }
     }
+
+    public static String getMinSec(long durationMs) {
+        long seconds = durationMs / 1000;
+        long minutes = seconds / 60;
+        long remainingSeconds = seconds % 60;
+
+        if (minutes == 0 && remainingSeconds == 0) {
+            return String.format("%d ms", durationMs);
+        }
+
+        return String.format("%d min %d sec", minutes, remainingSeconds);
+    }
+
+    abstract protected Map<String, Long> getTableColumnToIdMap(ResultSet resultSet) throws SQLException;
 
-    abstract protected byte[] createPayload(ResultSet resultSet, PubtransTableProtos.Common common, PubtransTableProtos.DOITripInfo tripInfo) throws SQLException;
+    abstract protected byte[] createPayload(
+            PubtransTableProtos.Common common, Map<String, Long> columnToIdMap,
+            PubtransTableProtos.DOITripInfo tripInfo) throws SQLException;
 
     abstract protected String getTimetabledDateTimeColumnName();
 
     abstract protected TransitdataSchema getSchema();
 
-    public Collection<TypedMessageBuilder<byte[]>> handleResultSet(ResultSet resultSet) throws SQLException {
+    public Collection<TypedMessageBuilder<byte[]>> handleResultSet(
+            ResultSet resultSet, PreparedStatement statement, long queryStartTime)
+            throws SQLException {
         List<TypedMessageBuilder<byte[]>> messageBuilderQueue = new ArrayList<>();
 
         long tempTimeStamp = getLastModifiedTimeStamp();
 
         int count = 0;
         int metroTripCount = 0;
         Set<String> metroRouteIds = new HashSet<>();
+        List<QueryResultItem> queryResultItems = new ArrayList<>();
+        long queryDuration = -1L;
+        long resultHandlerDuration = -1L;
+        long queryAndResultHandlerDuration = -1L;
 
         while (resultSet.next()) {
             count++;
-
+            
             PubtransTableProtos.Common common = parseCommon(resultSet);
             final long eventTimestampUtcMs = common.getLastModifiedUtcDateTimeMs();
-
+            
             final long delay = System.currentTimeMillis() - eventTimestampUtcMs;
             log.debug("Delay between current time and estimate publish time is {} ms", delay);
-
+            
             final String key = resultSet.getString("IsOnDatedVehicleJourneyId") + resultSet.getString("JourneyPatternSequenceNumber");
-            final long dvjId = common.getIsOnDatedVehicleJourneyId();
-            final long scheduledJppId = common.getIsTimetabledAtJourneyPatternPointGid();
-            final long targetedJppId = common.getIsTargetedAtJourneyPatternPointGid();
+            final Map<String, Long> columnToIdMap = getTableColumnToIdMap(resultSet);
+            queryResultItems.add(new QueryResultItem(common, key, eventTimestampUtcMs, columnToIdMap));
+        }
+
+        PubtransConnector.closeQuery(resultSet, statement);
+        long queryEndTime = System.currentTimeMillis();
+        queryDuration = queryEndTime - queryStartTime;
+
+        for (QueryResultItem queryResultItem : queryResultItems) {
+            final long dvjId = queryResultItem.common.getIsOnDatedVehicleJourneyId();
+            final long scheduledJppId = queryResultItem.common.getIsTimetabledAtJourneyPatternPointGid();
+            final long targetedJppId = queryResultItem.common.getIsTargetedAtJourneyPatternPointGid();
 
             Optional<PubtransTableProtos.DOITripInfo> maybeTripInfo = getTripInfo(dvjId, scheduledJppId, targetedJppId);
             if (maybeTripInfo.isEmpty()) {
@@ -102,21 +136,28 @@ public Collection<TypedMessageBuilder<byte[]>> handleResultSet(ResultSet resultS
                     metroTripCount++;
                     metroRouteIds.add(tripInfo.getRouteId());
                 } else {
-                    final byte[] data = createPayload(resultSet, common, tripInfo);
-                    TypedMessageBuilder<byte[]> msgBuilder = createMessage(key, eventTimestampUtcMs, dvjId, data, getSchema());
+                    final byte[] data = createPayload(queryResultItem.common, queryResultItem.columnToIdMap, tripInfo);
+                    TypedMessageBuilder<byte[]> msgBuilder = createMessage(queryResultItem.key,
+                            queryResultItem.eventTimestampUtcMs, dvjId, data, getSchema());
                     messageBuilderQueue.add(msgBuilder);
                 }
             }
 
             //Update latest ts for next round
-            if (eventTimestampUtcMs > tempTimeStamp) {
-                tempTimeStamp = eventTimestampUtcMs;
+            if (queryResultItem.eventTimestampUtcMs > tempTimeStamp) {
+                tempTimeStamp = queryResultItem.eventTimestampUtcMs;
             }
         }
-
-        log.info("{} rows processed from the result set. {} rows skipped with metro trips (route ids: {})",
-                count, metroTripCount, metroRouteIds);
-
+
+        long endTime = System.currentTimeMillis();
+        resultHandlerDuration = endTime - queryEndTime;
+        queryAndResultHandlerDuration = endTime - queryStartTime;
+
+        log.info("{} rows processed from the result set. {} rows skipped with metro trips (route ids: {}). "
+                        + "Operation took {} (db query took {}, handling results took {})",
+                count, metroTripCount, metroRouteIds,
+                getMinSec(queryAndResultHandlerDuration), getMinSec(queryDuration), getMinSec(resultHandlerDuration));
+
         setLastModifiedTimeStamp(tempTimeStamp);
 
         return messageBuilderQueue;

src/test/java/fi/hsl/transitdata/pulsarpubtransconnect/PubtransTableHandlerTest.java

@@ -39,5 +39,18 @@ public void testDaylightSavings() {
         assertEquals(summerTime, summerTimeWithDaylightSavings);
         assertEquals(summerTimeWithDaylightSavings, winterTime + 1);
     }
-
+
+    @Test
+    public void testConvertMillisecondsToMinutesAndSeconds() {
+        assertEquals("0 ms", PubtransTableHandler.getMinSec(0));
+        assertEquals("11 ms", PubtransTableHandler.getMinSec(11));
+        assertEquals("500 ms", PubtransTableHandler.getMinSec(500));
+        assertEquals("0 min 1 sec", PubtransTableHandler.getMinSec(1000));
+        assertEquals("0 min 59 sec", PubtransTableHandler.getMinSec(59000));
+        assertEquals("1 min 0 sec", PubtransTableHandler.getMinSec(60000));
+        assertEquals("1 min 1 sec", PubtransTableHandler.getMinSec(61000));
+        assertEquals("2 min 0 sec", PubtransTableHandler.getMinSec(120000));
+        assertEquals("2 min 1 sec", PubtransTableHandler.getMinSec(121000));
+        assertEquals("59 min 59 sec", PubtransTableHandler.getMinSec(3599000));
+    }
 }