Batching Kafka Connect JDBC Postgres Dialect

BatchingPostgreSqlDatabaseDialect.java

package your.company.here;

import io.confluent.connect.jdbc.dialect.DatabaseDialect;
import io.confluent.connect.jdbc.dialect.DatabaseDialectProvider;
import io.confluent.connect.jdbc.dialect.PostgreSqlDatabaseDialect;
import org.apache.kafka.common.config.AbstractConfig;

import java.sql.Connection;
import java.sql.SQLException;

public class BatchingPostgreSqlDatabaseDialect extends PostgreSqlDatabaseDialect {

    /**
     * The provider for {@link BatchingPostgreSqlDatabaseDialect}.
     */
    public static class Provider extends DatabaseDialectProvider.SubprotocolBasedProvider {
        public Provider() {
            super(BatchingPostgreSqlDatabaseDialect.class.getSimpleName(), "postgresql");
        }

        @Override
        public int score(final JdbcUrlInfo urlInfo) {
            return super.score(urlInfo) + 100; // force this version to be used
        }

        @Override
        public DatabaseDialect create(final AbstractConfig config) {
            return new BatchingPostgreSqlDatabaseDialect(config);
        }
    }

    /**
     * Create a new dialect instance with the given connector configuration.
     *
     * @param config the connector configuration; may not be null
     */
    public BatchingPostgreSqlDatabaseDialect(final AbstractConfig config) {
        super(config);
        System.out.println("Created dialect with config " + config);
    }

    @Override
    public Connection getConnection() throws SQLException {
        final Connection connection = super.getConnection();
        // disable autocommit so that row limits will be honored when fetching data
        // see https://github.com/confluentinc/kafka-connect-jdbc/issues/34
        connection.setAutoCommit(false);
        return connection;
    }
}

io.confluent.connect.jdbc.dialect.DatabaseDialectProvider

your.company.here.BatchingPostgreSqlDatabaseDialect$Provider