<commit_before>package io.confluent.connect.jdbc.sink.writer;

import com.google.common.collect.Iterators;

import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.sink.SinkRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Set;

import io.confluent.connect.jdbc.sink.ConnectionProvider;
import io.confluent.connect.jdbc.sink.Database;
import io.confluent.connect.jdbc.sink.SinkRecordField;
import io.confluent.connect.jdbc.sink.avro.AvroToDbConverter;
import io.confluent.connect.jdbc.sink.binders.PreparedStatementBinder;
import io.confluent.connect.jdbc.sink.common.DatabaseMetadata;
import io.confluent.connect.jdbc.sink.common.DatabaseMetadataProvider;
import io.confluent.connect.jdbc.sink.common.ParameterValidator;
import io.confluent.connect.jdbc.sink.config.FieldAlias;
import io.confluent.connect.jdbc.sink.config.FieldsMappings;
import io.confluent.connect.jdbc.sink.config.JdbcSinkSettings;
import io.confluent.connect.jdbc.sink.dialect.DbDialect;
import io.confluent.kafka.schemaregistry.client.rest.RestService;
import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException;

/**
 * Responsible for taking a sequence of SinkRecord and writing them to the database
 */
public final class JdbcDbWriter {
  private static final Logger logger = LoggerFactory.getLogger(JdbcDbWriter.class);

  private final PreparedStatementContextIterable statementBuilder;
  private final ErrorHandlingPolicy errorHandlingPolicy;
  private final Database database;
  private int retries;
  private Date lastError;
  private String lastErrorMessage;
  private final int maxRetries;
  private Connection connection = null;

  //provides connection pooling
  private final ConnectionProvider connectionProvider;

  /**
   * @param connectionProvider - The database connection provider
   * @param statementBuilder - Returns a sequence of PreparedStatement to process
   * @param errorHandlingPolicy - An instance of the error handling approach
   * @param database - Contains the database metadata (tables and their columns)
   * @param retries - Number of attempts to run when a SQLException occurs
   */
  public JdbcDbWriter(final ConnectionProvider connectionProvider,
                      final PreparedStatementContextIterable statementBuilder,
                      final ErrorHandlingPolicy errorHandlingPolicy,
                      final Database database,
                      final int retries) {
    ParameterValidator.notNull(connectionProvider, "connectionProvider");
    ParameterValidator.notNull(statementBuilder, "statementBuilder");
    ParameterValidator.notNull(database, "databaseChangesExecutor");

    this.connectionProvider = connectionProvider;
    this.statementBuilder = statementBuilder;
    this.errorHandlingPolicy = errorHandlingPolicy;
    this.database = database;
    this.retries = retries;
    this.maxRetries = retries;
  }

  /**
   * Writes the given records to the database
   *
   * @param records - The sequence of records to insert
   */
  public void write(final Collection<SinkRecord> records) {
    if (records.isEmpty()) {
      logger.warn("Received empty sequence of SinkRecord");
    } else {

      final Iterator<PreparedStatementContext> iterator = statementBuilder.iterator(records);
      try {
        //initialize a new connection instance if we haven't done it before
        if (connection == null) {
          connection = connectionProvider.getConnection();
        } else if (!connection.isValid(3000)) {
          //check the connection is still valid
          logger.warn("The database connection is not valid. Reconnecting");
          closeConnectionQuietly();
          connection = connectionProvider.getConnection();
        }
        //begin transaction
        connection.setAutoCommit(false);

        int totalRecords = 0;

        //::hasNext can say it has items but ::next can actually throw NoSuchElementException
        //see the iterator where because of filtering we might not actually have data to process
        try {
          while (iterator.hasNext()) {

            final PreparedStatementContext statementContext = iterator.next();
            final PreparedStatementData statementData = statementContext.getPreparedStatementData();

            //handle possible database changes (new tables, new columns)
            database.update(statementContext.getTablesToColumnsMap(), connection);

            PreparedStatement statement = null;
            try {
              final String sql = statementData.getSql();
              logger.debug("Executing SQL: {}", sql);
              statement = connection.prepareStatement(sql);
              for (Iterable<PreparedStatementBinder> entryBinders : statementData.getBinders()) {
                PreparedStatementBindData.apply(statement, entryBinders);
                statement.addBatch();
                totalRecords++;
              }
              statement.executeBatch();

            } finally {
              if (statement != null) {
                statement.close();
              }
            }
          }
        } catch (NoSuchElementException ex) {
          //yes we can end up here; but it is not an issue
          logger.warn(ex.getMessage());
        }
        //commit the transaction
        connection.commit();

        logger.info("Wrote " + totalRecords + " to the database.");
        if (maxRetries != retries) {
          retries = maxRetries;
          SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS'Z'");
          logger.info("Recovered from error '{}' ({})", lastErrorMessage, formatter.format(lastError));<commit_msg>try-with-resources for statement<commit_after>package io.confluent.connect.jdbc.sink.writer;

import com.google.common.collect.Iterators;

import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.sink.SinkRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Set;

import io.confluent.connect.jdbc.sink.ConnectionProvider;
import io.confluent.connect.jdbc.sink.Database;
import io.confluent.connect.jdbc.sink.SinkRecordField;
import io.confluent.connect.jdbc.sink.avro.AvroToDbConverter;
import io.confluent.connect.jdbc.sink.binders.PreparedStatementBinder;
import io.confluent.connect.jdbc.sink.common.DatabaseMetadata;
import io.confluent.connect.jdbc.sink.common.DatabaseMetadataProvider;
import io.confluent.connect.jdbc.sink.common.ParameterValidator;
import io.confluent.connect.jdbc.sink.config.FieldAlias;
import io.confluent.connect.jdbc.sink.config.FieldsMappings;
import io.confluent.connect.jdbc.sink.config.JdbcSinkSettings;
import io.confluent.connect.jdbc.sink.dialect.DbDialect;
import io.confluent.kafka.schemaregistry.client.rest.RestService;
import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException;

/**
 * Responsible for taking a sequence of SinkRecord and writing them to the database
 */
public final class JdbcDbWriter {
  private static final Logger logger = LoggerFactory.getLogger(JdbcDbWriter.class);

  private final PreparedStatementContextIterable statementBuilder;
  private final ErrorHandlingPolicy errorHandlingPolicy;
  private final Database database;
  private int retries;
  private Date lastError;
  private String lastErrorMessage;
  private final int maxRetries;
  private Connection connection = null;

  //provides connection pooling
  private final ConnectionProvider connectionProvider;

  /**
   * @param connectionProvider - The database connection provider
   * @param statementBuilder - Returns a sequence of PreparedStatement to process
   * @param errorHandlingPolicy - An instance of the error handling approach
   * @param database - Contains the database metadata (tables and their columns)
   * @param retries - Number of attempts to run when a SQLException occurs
   */
  public JdbcDbWriter(final ConnectionProvider connectionProvider,
                      final PreparedStatementContextIterable statementBuilder,
                      final ErrorHandlingPolicy errorHandlingPolicy,
                      final Database database,
                      final int retries) {
    ParameterValidator.notNull(connectionProvider, "connectionProvider");
    ParameterValidator.notNull(statementBuilder, "statementBuilder");
    ParameterValidator.notNull(database, "databaseChangesExecutor");

    this.connectionProvider = connectionProvider;
    this.statementBuilder = statementBuilder;
    this.errorHandlingPolicy = errorHandlingPolicy;
    this.database = database;
    this.retries = retries;
    this.maxRetries = retries;
  }

  /**
   * Writes the given records to the database
   *
   * @param records - The sequence of records to insert
   */
  public void write(final Collection<SinkRecord> records) {
    if (records.isEmpty()) {
      logger.warn("Received empty sequence of SinkRecord");
    } else {

      final Iterator<PreparedStatementContext> iterator = statementBuilder.iterator(records);
      try {
        //initialize a new connection instance if we haven't done it before
        if (connection == null) {
          connection = connectionProvider.getConnection();
        } else if (!connection.isValid(3000)) {
          //check the connection is still valid
          logger.warn("The database connection is not valid. Reconnecting");
          closeConnectionQuietly();
          connection = connectionProvider.getConnection();
        }
        //begin transaction
        connection.setAutoCommit(false);

        int totalRecords = 0;

        //::hasNext can say it has items but ::next can actually throw NoSuchElementException
        //see the iterator where because of filtering we might not actually have data to process
        try {
          while (iterator.hasNext()) {

            final PreparedStatementContext statementContext = iterator.next();
            final PreparedStatementData statementData = statementContext.getPreparedStatementData();

            //handle possible database changes (new tables, new columns)
            database.update(statementContext.getTablesToColumnsMap(), connection);

            final String sql = statementData.getSql();
            logger.debug("Executing SQL: {}", sql);
            try (PreparedStatement statement = connection.prepareStatement(sql)) {
              for (Iterable<PreparedStatementBinder> entryBinders : statementData.getBinders()) {
                PreparedStatementBindData.apply(statement, entryBinders);
                statement.addBatch();
                totalRecords++;
              }
              statement.executeBatch();
            }
          }
        } catch (NoSuchElementException ex) {
          //yes we can end up here; but it is not an issue
          logger.warn(ex.getMessage());
        }
        //commit the transaction
        connection.commit();

        logger.info("Wrote " + totalRecords + " to the database.");
        if (maxRetries != retries) {
          retries = maxRetries;
          SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS'Z'");
          logger.info("Recovered from error '{}' ({})", lastErrorMessage, formatter.format(lastError));
{'fraction_non_alphanumeric': 0.0859563159206046, 'fraction_numerical': 0.0011626941283946516, 'mean_word_length': 3.344155844155844, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3751762', 'n_tokens_mistral': 3149, 'n_tokens_neox': 3076, 'n_words': 788}