ListDatabaseTables processor doesn't close ResultSets
This closes
@@ -230,7 +230,7 @@ public class ListDatabaseTables extends AbstractProcessor {
try (final Connection con = dbcpService.getConnection(Collections.emptyMap())) {
DatabaseMetaData dbMetaData = con.getMetaData();
- ResultSet rs = dbMetaData.getTables(catalog, schemaPattern, tableNamePattern, tableTypes);
+ try (ResultSet rs = dbMetaData.getTables(catalog, schemaPattern, tableNamePattern, tableTypes)) {
while (rs.next()) {
final String tableCatalog = rs.getString(1);
final String tableSchema = rs.getString(2);
@@ -252,13 +252,15 @@ public class ListDatabaseTables extends AbstractProcessor {
if (!StringUtils.isEmpty(lastTimestampForTable)) {
lastRefreshed = Long.parseLong(lastTimestampForTable);
}
- if (lastRefreshed == -1 || (refreshInterval > 0 && currentTime >= (lastRefreshed + refreshInterval))) {
+ if (lastRefreshed == -1 || (refreshInterval > 0 && currentTime >= (lastRefreshed
+ + refreshInterval))) {
stateMapProperties.remove(lastTimestampForTable);
} else {
refreshTable = false;
}
} catch (final NumberFormatException nfe) {
- getLogger().error("Failed to retrieve observed last table fetches from the State Manager. Will not perform "
+ getLogger().error(
+ "Failed to retrieve observed last table fetches from the State Manager. Will not perform "
+ "query until this is accomplished.", nfe);
context.yield();
return;
@@ -271,9 +273,12 @@ public class ListDatabaseTables extends AbstractProcessor {
final String countQuery = "SELECT COUNT(1) FROM " + fqn;
logger.debug("Executing query: {}", new Object[] {countQuery});
- ResultSet countResult = st.executeQuery(countQuery);
+ try (ResultSet countResult = st.executeQuery(countQuery)) {
if (countResult.next()) {
- flowFile = session.putAttribute(flowFile, DB_TABLE_COUNT, Long.toString(countResult.getLong(1)));
+ flowFile = session
+ .putAttribute(flowFile, DB_TABLE_COUNT,
+ Long.toString(countResult.getLong(1)));
+ }
}
} catch (SQLException se) {
logger.error("Couldn't get row count for {}", new Object[] {fqn});
@@ -305,6 +310,7 @@ public class ListDatabaseTables extends AbstractProcessor {
stateMapProperties.put(fqn, Long.toString(System.currentTimeMillis()));
}
}
+ }
// Update the timestamps for listed tables
if (stateMap.getVersion() == -1) {
stateManager.setState(stateMapProperties, Scope.CLUSTER);
