Merge branch 'STORM-2028' of https://github.com/rahuljain373/storm into STORM-2028-merge
Project: http://git-wip-us.apache.org/repos/asf/storm/repo Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/b48a8a1a Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/b48a8a1a Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/b48a8a1a Branch: refs/heads/master Commit: b48a8a1a64404e98d4c408dd68caaa5f27e450f3 Parents: 4c9a864 921635a Author: Jungtaek Lim <[email protected]> Authored: Wed Aug 9 13:43:57 2017 +0900 Committer: Jungtaek Lim <[email protected]> Committed: Wed Aug 9 13:43:57 2017 +0900 ---------------------------------------------------------------------- .../apache/storm/jdbc/common/JdbcClient.java | 42 ++++++++++++++----- .../storm/jdbc/common/JdbcClientTest.java | 43 ++++++++++++++++++++ 2 files changed, 75 insertions(+), 10 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/storm/blob/b48a8a1a/external/storm-jdbc/src/main/java/org/apache/storm/jdbc/common/JdbcClient.java ---------------------------------------------------------------------- diff --cc external/storm-jdbc/src/main/java/org/apache/storm/jdbc/common/JdbcClient.java index 00f8a6e,e29edd5..f591236 --- a/external/storm-jdbc/src/main/java/org/apache/storm/jdbc/common/JdbcClient.java +++ b/external/storm-jdbc/src/main/java/org/apache/storm/jdbc/common/JdbcClient.java @@@ -107,57 -110,58 +111,60 @@@ public class JdbcClient Connection connection = null; try { connection = connectionProvider.getConnection(); - PreparedStatement preparedStatement = connection.prepareStatement(sqlQuery); - if(queryTimeoutSecs > 0) { - preparedStatement.setQueryTimeout(queryTimeoutSecs); - } - setPreparedStatementParams(preparedStatement, queryParams); - ResultSet resultSet = preparedStatement.executeQuery(); - List<List<Column>> rows = Lists.newArrayList(); - while(resultSet.next()){ - ResultSetMetaData metaData = resultSet.getMetaData(); - int columnCount = metaData.getColumnCount(); - List<Column> row = Lists.newArrayList(); - for(int i=1 ; i <= columnCount; i++) { - String columnLabel = metaData.getColumnLabel(i); - int columnType = metaData.getColumnType(i); - Class columnJavaType = Util.getJavaType(columnType); - if (columnJavaType.equals(String.class)) { - row.add(new Column<String>(columnLabel, resultSet.getString(columnLabel), columnType)); - } else if (columnJavaType.equals(Integer.class)) { - row.add(new Column<Integer>(columnLabel, resultSet.getInt(columnLabel), columnType)); - } else if (columnJavaType.equals(Double.class)) { - row.add(new Column<Double>(columnLabel, resultSet.getDouble(columnLabel), columnType)); - } else if (columnJavaType.equals(Float.class)) { - row.add(new Column<Float>(columnLabel, resultSet.getFloat(columnLabel), columnType)); - } else if (columnJavaType.equals(Short.class)) { - row.add(new Column<Short>(columnLabel, resultSet.getShort(columnLabel), columnType)); - } else if (columnJavaType.equals(Boolean.class)) { - row.add(new Column<Boolean>(columnLabel, resultSet.getBoolean(columnLabel), columnType)); - } else if (columnJavaType.equals(byte[].class)) { - row.add(new Column<byte[]>(columnLabel, resultSet.getBytes(columnLabel), columnType)); - } else if (columnJavaType.equals(Long.class)) { - row.add(new Column<Long>(columnLabel, resultSet.getLong(columnLabel), columnType)); - } else if (columnJavaType.equals(Date.class)) { - row.add(new Column<Date>(columnLabel, resultSet.getDate(columnLabel), columnType)); - } else if (columnJavaType.equals(Time.class)) { - row.add(new Column<Time>(columnLabel, resultSet.getTime(columnLabel), columnType)); - } else if (columnJavaType.equals(Timestamp.class)) { - row.add(new Column<Timestamp>(columnLabel, resultSet.getTimestamp(columnLabel), columnType)); - } else { - throw new RuntimeException("type = " + columnType + " for column " + columnLabel + " not supported."); + try (PreparedStatement preparedStatement = connection.prepareStatement(sqlQuery)) { + if (queryTimeoutSecs > 0) { + preparedStatement.setQueryTimeout(queryTimeoutSecs); + } + setPreparedStatementParams(preparedStatement, queryParams); + try (ResultSet resultSet = preparedStatement.executeQuery()) { + List<List<Column>> rows = Lists.newArrayList(); + while (resultSet.next()) { + ResultSetMetaData metaData = resultSet.getMetaData(); + int columnCount = metaData.getColumnCount(); + List<Column> row = Lists.newArrayList(); + for (int i = 1; i <= columnCount; i++) { + String columnLabel = metaData.getColumnLabel(i); + int columnType = metaData.getColumnType(i); + Class columnJavaType = Util.getJavaType(columnType); + if (columnJavaType.equals(String.class)) { + row.add(new Column<String>(columnLabel, resultSet.getString(columnLabel), columnType)); + } else if (columnJavaType.equals(Integer.class)) { + row.add(new Column<Integer>(columnLabel, resultSet.getInt(columnLabel), columnType)); + } else if (columnJavaType.equals(Double.class)) { + row.add(new Column<Double>(columnLabel, resultSet.getDouble(columnLabel), columnType)); + } else if (columnJavaType.equals(Float.class)) { + row.add(new Column<Float>(columnLabel, resultSet.getFloat(columnLabel), columnType)); + } else if (columnJavaType.equals(Short.class)) { + row.add(new Column<Short>(columnLabel, resultSet.getShort(columnLabel), columnType)); + } else if (columnJavaType.equals(Boolean.class)) { + row.add(new Column<Boolean>(columnLabel, resultSet.getBoolean(columnLabel), columnType)); + } else if (columnJavaType.equals(byte[].class)) { + row.add(new Column<byte[]>(columnLabel, resultSet.getBytes(columnLabel), columnType)); + } else if (columnJavaType.equals(Long.class)) { + row.add(new Column<Long>(columnLabel, resultSet.getLong(columnLabel), columnType)); + } else if (columnJavaType.equals(Date.class)) { + row.add(new Column<Date>(columnLabel, resultSet.getDate(columnLabel), columnType)); + } else if (columnJavaType.equals(Time.class)) { + row.add(new Column<Time>(columnLabel, resultSet.getTime(columnLabel), columnType)); + } else if (columnJavaType.equals(Timestamp.class)) { + row.add(new Column<Timestamp>(columnLabel, resultSet.getTimestamp(columnLabel), columnType)); + } else { + throw new RuntimeException("type = " + columnType + " for column " + columnLabel + " not supported."); + } + } + rows.add(row); } + return rows; } - rows.add(row); } - return rows; } catch (SQLException e) { - throw new RuntimeException("Failed to execute select query " + sqlQuery, e); + selectException = new RuntimeException("Failed to execute select query " + sqlQuery, e); + } catch (RuntimeException e) { + selectException = e; } finally { - closeConnection(connection); + closeConnection(connection, selectException); } + return null; } public List<Column> getColumnSchema(String tableName) { @@@ -182,13 -191,14 +194,15 @@@ Connection connection = null; try { connection = connectionProvider.getConnection(); - Statement statement = connection.createStatement(); - statement.execute(sql); + try (Statement statement = connection.createStatement()) { + statement.execute(sql); + } } catch (SQLException e) { - throw new RuntimeException("Failed to execute SQL", e); + execException = new RuntimeException("Failed to execute SQL", e); + } catch (RuntimeException e) { + execException = e; } finally { - closeConnection(connection); + closeConnection(connection, execException); } }
