refactor: Update folder structure and sort import

Signed-off-by: Joel Hanson <[email protected]>

Author: Joel-hanson

src/main/java/com/ibm/eventstreams/connect/jdbcsink/JDBCSinkConfig.java

@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2020 IBM Corporation
+ * Copyright 2020, 2023 IBM Corporation
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,11 +18,11 @@
 
 package com.ibm.eventstreams.connect.jdbcsink;
 
+import java.util.Map;
+
 import org.apache.kafka.common.config.AbstractConfig;
 import org.apache.kafka.common.config.ConfigDef;
 
-import java.util.Map;
-
 public class JDBCSinkConfig extends AbstractConfig {
 
     private static final String CONFIG_CONNECTION_GROUP = "Connection";
@@ -49,7 +49,7 @@ public class JDBCSinkConfig extends AbstractConfig {
 
     public static final String CONFIG_NAME_INSERT_MODE_DATABASELEVEL = "insert.mode.databaselevel";
     private static final String CONFIG_DOCUMENTATION_INSERT_MODE_DATABASELEVEL = "The insertion mode to use (ex: insert, upsert, or update).";
-    private static final String CONFIG_DISPLAY_INSERT_MODE_DATABASELEVEL  = "Insert mode database level";
+    private static final String CONFIG_DISPLAY_INSERT_MODE_DATABASELEVEL = "Insert mode database level";
 
     public static ConfigDef config() {
         ConfigDef config = new ConfigDef();

src/main/java/com/ibm/eventstreams/connect/jdbcsink/JDBCSinkConnector.java

@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2020 IBM Corporation
+ * Copyright 2020, 2023 IBM Corporation
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,15 +18,15 @@
 
 package com.ibm.eventstreams.connect.jdbcsink;
 
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
 import org.apache.kafka.common.config.Config;
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.connect.connector.Task;
 import org.apache.kafka.connect.sink.SinkConnector;
 
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-
 public class JDBCSinkConnector extends SinkConnector {
     // TODO: check with ibm-messaging about externalizing snapshot version
     public static String VERSION = "0.0.3-SNAPSHOT";
@@ -38,24 +38,28 @@ public class JDBCSinkConnector extends SinkConnector {
      *
      * @return the version, formatted as a String
      */
-    @Override public String version() {
+    @Override
+    public String version() {
         return VERSION;
     }
 
     /**
-     * Start this Connector. This method will only be called on a clean Connector, i.e. it has
-     * either just been instantiated and initialized or {@link #stop()} has been invoked.
+     * Start this Connector. This method will only be called on a clean Connector,
+     * i.e. it has either just been instantiated and initialized or {@link #stop()}
+     * has been invoked.
      *
      * @param props configuration settings
      */
-    @Override public void start(Map<String, String> props) {
+    @Override
+    public void start(Map<String, String> props) {
         this.props = props;
     }
 
     /**
      * Returns the Task implementation for this Connector.
      */
-    @Override public Class<? extends Task> taskClass() {
+    @Override
+    public Class<? extends Task> taskClass() {
         return JDBCSinkTask.class;
     }
 
@@ -66,33 +70,39 @@ public class JDBCSinkConnector extends SinkConnector {
      * @param maxTasks maximum number of configurations to generate
      * @return configurations for Tasks
      */
-    @Override public List<Map<String, String>> taskConfigs(int maxTasks) {
+    @Override
+    public List<Map<String, String>> taskConfigs(int maxTasks) {
         return Collections.nCopies(maxTasks, props);
     }
 
     /**
      * Stop this connector.
      */
-    @Override public void stop() {
+    @Override
+    public void stop() {
 
     }
 
     /**
      * Define the configuration for the connector.
+     *
      * @return The ConfigDef for this connector.
      */
-    @Override public ConfigDef config() {
+    @Override
+    public ConfigDef config() {
         return JDBCSinkConfig.config();
     }
 
     /**
-     * Provides a default validation implementation which returns a list of allowed configurations
-     * together with configuration errors and recommended values for each configuration.
+     * Provides a default validation implementation which returns a list of allowed
+     * configurations together with configuration errors and recommended values for
+     * each configuration.
      *
      * @param connectorConfigs connector configuration values
      * @return list of allowed configurations
      */
-    @Override public Config validate(Map<String, String> connectorConfigs) {
+    @Override
+    public Config validate(Map<String, String> connectorConfigs) {
         return super.validate(connectorConfigs);
     }
 

src/main/java/com/ibm/eventstreams/connect/jdbcsink/JDBCSinkTask.java

@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2020 IBM Corporation
+ * Copyright 2020, 2023 IBM Corporation
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,75 +18,87 @@
 
 package com.ibm.eventstreams.connect.jdbcsink;
 
-import com.ibm.eventstreams.connect.jdbcsink.database.DatabaseFactory;
-import com.ibm.eventstreams.connect.jdbcsink.database.IDatabase;
+import java.sql.SQLException;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Collection;
+import java.util.Map;
+
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.connect.connector.Connector;
+import org.apache.kafka.connect.errors.ConnectException;
 import org.apache.kafka.connect.sink.SinkRecord;
 import org.apache.kafka.connect.sink.SinkTask;
 import org.apache.kafka.connect.sink.SinkTaskContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.sql.SQLException;
-import java.time.Duration;
-import java.time.Instant;
-import java.util.Collection;
-import java.util.Map;
+import com.ibm.eventstreams.connect.jdbcsink.database.DatabaseFactory;
+import com.ibm.eventstreams.connect.jdbcsink.database.IDatabase;
 
 public class JDBCSinkTask extends SinkTask {
     private static final Logger logger = LoggerFactory.getLogger(JDBCSinkTask.class);
     private static final String classname = JDBCSinkTask.class.getName();
 
     // TODO: needs to be generic and incorporate other database types
-    //  needs an interface
+    // needs an interface
     private JDBCSinkConfig config;
 
     public IDatabase database;
 
     int remainingRetries; // init max retries via config.maxRetries ...
 
     /**
-     * Start the Task. This should handle any configuration parsing and one-time setup of the task.
+     * Start the Task. This should handle any configuration parsing and one-time
+     * setup of the task.
+     *
      * @param props initial configuration
      */
-    @Override public void start(Map<String, String> props) {
+    @Override
+    public void start(Map<String, String> props) {
         logger.trace("[{}] Entry {}.start, props={}", Thread.currentThread().getId(), classname, props);
         this.config = new JDBCSinkConfig(props);
 
-        DatabaseFactory databaseFactory = new DatabaseFactory();
+        DatabaseFactory databaseFactory = getDatabaseFactory();
         try {
             this.database = databaseFactory.makeDatabase(this.config);
         } catch (Exception e) {
             logger.error("Failed to build the database {} ", e);
-            e.printStackTrace();
-            throw e;
+            throw new ConnectException(e);
         }
 
         logger.trace("[{}]  Exit {}.start", Thread.currentThread().getId(), classname);
     }
 
+    protected DatabaseFactory getDatabaseFactory() {
+        DatabaseFactory databaseFactory = new DatabaseFactory();
+        return databaseFactory;
+    }
+
     /**
      * Put the records in the sink.
      *
-     * If this operation fails, the SinkTask may throw a {@link org.apache.kafka.connect.errors.RetriableException} to
-     * indicate that the framework should attempt to retry the same call again. Other exceptions will cause the task to
-     * be stopped immediately. {@link SinkTaskContext#timeout(long)} can be used to set the maximum time before the
-     * batch will be retried.
+     * If this operation fails, the SinkTask may throw a
+     * {@link org.apache.kafka.connect.errors.RetriableException} to indicate that
+     * the framework should attempt to retry the same call again. Other exceptions
+     * will cause the task to be stopped immediately.
+     * {@link SinkTaskContext#timeout(long)} can be used to set the maximum time
+     * before the batch will be retried.
      *
      * @param records the set of records to send
      */
-    @Override public void put(Collection<SinkRecord> records) {
+    @Override
+    public void put(Collection<SinkRecord> records) {
+        logger.trace("[{}] Entry {}.put", Thread.currentThread().getId(), classname);
         if (records.isEmpty()) {
             return;
         }
 
         final SinkRecord first = records.iterator().next();
         final int recordsCount = records.size();
         logger.info("Received {} records. First record kafka coordinates:({}-{}-{}). Writing them to the database...",
-                recordsCount, first.topic(), first.kafkaPartition(), first.kafkaOffset()
-        );
+                recordsCount, first.topic(), first.kafkaPartition(), first.kafkaOffset());
 
         final String tableName = config.getString(JDBCSinkConfig.CONFIG_NAME_TABLE_NAME_FORMAT);
 
@@ -96,28 +108,37 @@ public class JDBCSinkTask extends SinkTask {
             this.database.getWriter().insert(tableName, records);
             logger.info(String.format("%d RECORDS PROCESSED", records.size()));
             Instant finish = Instant.now();
-            long timeElapsed = Duration.between(start, finish).toMillis();  //in millis
-            logger.info(String.format("Processed '%d' records", records.size() ));
+            long timeElapsed = Duration.between(start, finish).toMillis(); // in millis
+            logger.info(String.format("Processed '%d' records", records.size()));
             logger.info(String.format("Total Execution time: %d", timeElapsed));
         } catch (SQLException error) {
             logger.error("Write of {} records failed, remainingRetries={}", recordsCount, remainingRetries, error);
-            // TODO: throw exception to cancel execution or retry?
+            throw new ConnectException(error);
+        } catch (final RuntimeException e) {
+            logger.error("Unexpected runtime exception: ", e);
+            throw e;
         }
+
+        logger.trace("[{}] Exit {}.put", Thread.currentThread().getId(), classname);
     }
 
-    @Override public void stop() {
+    @Override
+    public void stop() {
     }
 
-    @Override public void flush(Map<TopicPartition, OffsetAndMetadata> map) {
+    @Override
+    public void flush(Map<TopicPartition, OffsetAndMetadata> map) {
         // Not necessary
     }
 
     /**
-     * Get the version of this task. Usually this should be the same as the corresponding {@link Connector} class's version.
+     * Get the version of this task. Usually this should be the same as the
+     * corresponding {@link Connector} class's version.
      *
      * @return the version, formatted as a String
      */
-    @Override public String version() {
+    @Override
+    public String version() {
         return getClass().getPackage().getImplementationVersion();
     }
 }

src/main/java/com/ibm/eventstreams/connect/jdbcsink/database/DatabaseFactory.java

@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2020 IBM Corporation
+ * Copyright 2020, 2023 IBM Corporation
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,15 +18,16 @@
 
 package com.ibm.eventstreams.connect.jdbcsink.database;
 
+import java.beans.PropertyVetoException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import com.ibm.eventstreams.connect.jdbcsink.JDBCSinkConfig;
 import com.ibm.eventstreams.connect.jdbcsink.database.datasource.IDataSource;
 import com.ibm.eventstreams.connect.jdbcsink.database.datasource.PooledDataSource;
 import com.ibm.eventstreams.connect.jdbcsink.database.exception.DatabaseNotSupportedException;
 import com.ibm.eventstreams.connect.jdbcsink.database.exception.JdbcDriverClassNotFoundException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.beans.PropertyVetoException;
 
 public class DatabaseFactory {
     private static final Logger logger = LoggerFactory.getLogger(DatabaseFactory.class);
@@ -35,39 +36,53 @@ public IDatabase makeDatabase(JDBCSinkConfig config) {
 
         logger.warn("DatabaseFactory: makeDatabase");
 
-        String jdbcUrl = config.getString(JDBCSinkConfig.CONFIG_NAME_CONNECTION_URL);
+        DatabaseType databaseType = getDatabaseType(config);
+
+        String databaseDriver = getDatabaseDriver(databaseType);
+
+        IDataSource dataSource = getDataSource(config, databaseDriver);
 
+        return databaseType.create(dataSource);
+    }
+
+    private DatabaseType getDatabaseType(JDBCSinkConfig config) {
+        String jdbcUrl = config.getString(JDBCSinkConfig.CONFIG_NAME_CONNECTION_URL);
         DatabaseType databaseType = DatabaseType.fromJdbcUrl(jdbcUrl);
 
         if (databaseType == null) {
             throw new DatabaseNotSupportedException("Check " + jdbcUrl);
         }
+        return databaseType;
+    }
 
-        String databaseDriver = databaseType.getDriver();
-        try {
-            Class.forName(databaseDriver);
-        } catch (ClassNotFoundException cnf) {
-            logger.error(databaseType.name() + " JDBC driver not found", cnf);
-            throw new JdbcDriverClassNotFoundException(databaseType.name());
-        }
-
+    private IDataSource getDataSource(JDBCSinkConfig config, String databaseDriver) {
         final String username = config.getString(JDBCSinkConfig.CONFIG_NAME_CONNECTION_USER);
         final String password = config.getPassword(JDBCSinkConfig.CONFIG_NAME_CONNECTION_PASSWORD).value();
         final int poolSize = config.getInt(JDBCSinkConfig.CONFIG_NAME_CONNECTION_DS_POOL_SIZE);
+        String jdbcUrl = config.getString(JDBCSinkConfig.CONFIG_NAME_CONNECTION_URL);
 
         IDataSource dataSource = null;
         try {
             dataSource = new PooledDataSource.Builder(
                     username,
                     password,
                     jdbcUrl,
-                    databaseDriver
-            ).withInitialPoolSize(poolSize).build();
+                    databaseDriver).withInitialPoolSize(poolSize).build();
         } catch (PropertyVetoException e) {
             logger.error(e.toString());
         }
+        return dataSource;
+    }
 
-        return databaseType.create(dataSource);
+    private String getDatabaseDriver(DatabaseType databaseType) {
+        String databaseDriver = databaseType.getDriver();
+        try {
+            Class.forName(databaseDriver);
+        } catch (ClassNotFoundException cnf) {
+            logger.error(databaseType.name() + " JDBC driver not found", cnf);
+            throw new JdbcDriverClassNotFoundException(databaseType.name());
+        }
+        return databaseDriver;
     }
 
 }