StreamSets / Snowflake TIMESTAMP_TZ issue reproducer

Main.java

package test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.FileInputStream;
import java.util.Properties;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.Statement;

/*

Create and populate this Snowflake table:

create table test (tz TIMESTAMP_TZ);

insert into test values (current_timestamp)

*/


public class Main {

    public static void main(String[] args) {
        Connection conn = null;
        Statement stmt = null;
        ResultSet resultSet = null;
        String sql = "select * from mark_db.mark_schema.test";
        try {
            Properties props = getJdbcProperties();
            String jdbcUrl = createJdbcUrl(props);
            conn = createConnection(jdbcUrl, props);
            stmt = conn.createStatement();
            resultSet = stmt.executeQuery(sql);
            getResults(resultSet);
        } catch (Exception e) {
            System.out.println("-- Error -------------------");
            System.out.println(e);
            System.out.println("----------------------------");
        } finally {
            try {
                resultSet.close();
                stmt.close();
                conn.close();
            } catch (Exception e) {
                // swallow
            }
        }
    }

    private static void getResults(ResultSet resultSet) throws Exception {
        ResultSetMetaData metaData = resultSet.getMetaData();
        int columnCount = metaData.getColumnCount();
        while (resultSet.next()) {
            for (int i = 1; i <= columnCount; i++) {
                String columnName = metaData.getColumnName(i);
                String columnTypeName = metaData.getColumnTypeName(i);
                int columnTypeInt = metaData.getColumnType(i);
                Object columnValue;
                System.out.println("-----------------------------------------");
                System.out.println("Column Name: " + columnName);
                System.out.println("Column Type Name : " + columnTypeName);
                System.out.println("Column Type Integer : " + columnTypeInt);
                if (columnTypeInt == 2014){
                    System.out.println("------------------");
                    System.out.println("Trying to get Column using type 'java.sql.Timestamp.class'");
                    try{
                        columnValue = resultSet.getObject(i, java.sql.Timestamp.class);
                        System.out.println("Column value is: "  + columnValue);
                        System.out.println(" --> No exception is thrown when using the 'java.sql.Timestamp.class'");
                    } catch (Exception e){
                        e.printStackTrace();
                    }
                    System.out.println("------------------"); 
                    System.out.println("Trying to get Column using type 'net.snowflake.client.jdbc.SnowflakeTimestampWithTimezone.class'");
                    try {
                        columnValue = resultSet.getObject(i, net.snowflake.client.jdbc.SnowflakeTimestampWithTimezone.class);
                        System.out.println("Column value is: "  + columnValue);
                        System.out.println(" --> No exception is thrown when using the 'net.snowflake.client.jdbc.SnowflakeTimestampWithTimezone.class'");
                    } catch (Exception e){
                        e.printStackTrace();
                    }
                    System.out.println("------------------");  
                    System.out.println("Trying to get Column using type 'java.time.OffsetDateTime.class'");
                    try{
                        columnValue = resultSet.getObject(i, java.time.OffsetDateTime.class);
                        System.out.println("Column value is: "  + columnValue);
                    } catch (Exception e){
                        e.printStackTrace();
                    }   
                    System.out.println("------------------");
                }   
            }
        }
    }

    private static String createJdbcUrl(Properties props) {
        String url = props.getProperty("snowflake.url");
        String warehouse = props.getProperty("snowflake.warehouse");
        String database = props.getProperty("snowflake.database");
        String schema = props.getProperty("snowflake.schema");
        String role = props.getProperty("snowflake.role");
        return String.format("%s?warehouse=%s&db=%s&schema=%s&role=%s", url, warehouse, database, schema, role);

    }

    private static Connection createConnection(String jdbcUrl, Properties props) throws Exception {
        String user = props.getProperty("snowflake.user");
        String password = props.getProperty("snowflake.password");
        Class.forName("net.snowflake.client.jdbc.SnowflakeDriver");
        return DriverManager.getConnection(jdbcUrl, user, password);
    }

    private static Properties getJdbcProperties() throws Exception {
        Properties props = new Properties();
        FileInputStream fis = new FileInputStream("config.properties");
        props.load(fis);
        return props;
    }
}