@Override
    public void handleBatch(List<RecordChangeEvent<SourceRecord>> records, DebeziumEngine.RecordCommitter<RecordChangeEvent<SourceRecord>> committer) throws InterruptedException {
        LOG.info("New records received: {}", records.size());

        // batch operations
        Envelope.Operation oldOperation = null;
        String oldSinkTableName = null;

        for (RecordChangeEvent<SourceRecord> r : records) {
            SourceRecord record = r.record();
            if (record.value() != null) {
                LOG.debug(record);
                Envelope.Operation operation = Envelope.operationFor(record);

                if (operation != null) {
                    try {
                        // if the operation OR de sink table changes execute bath operations
                        if (batchPS != null) {
                            LOG.trace("batchPS es nulo");
                            LOG.trace("oldOperation: {}, newOperation: {}", oldOperation, operation);
                            LOG.trace("oldSinkTableName: {}, newSinkTableName: {}", oldSinkTableName, getSourceTableName(record));
                            if (
                                    (oldOperation != null && !oldOperation.equals(operation))
                                            ||
                                            (oldSinkTableName != null && !oldSinkTableName.equals(getSourceTableName(record)))
                            ) {

                              try {
                                  int[] rows = batchPS.executeBatch();
                                  this.getConnection().commit();
                                  LOG.info("Commited batch records. Rows affected: {}", rows.length);
                                  batchPS.close();
                                  batchPS = null;
                              }catch (Exception e){
                                  LOG.error("Error ejecutando el batchPS.executeBatch: {}",e.getMessage());
                                  LOG.error(e);
                                  // TODO Si hay un error en un batchPS se hace rollback en todo el batch y perdemos datos.
                                  // hay que ver como gestionar este error

                              }
                            }
                        }

                        switch (operation) {
                            case READ:
                                LOG.trace("Read event. Snapshoting - Merge");
                                doMerge(record);
                                break;
                            case CREATE:
                                LOG.trace("Create event. Insert");
                                doInsert(record);
                                break;
                            case DELETE:
                                LOG.trace("Delete event. Delete");
                                doDelete(record);
                                break;
                            case UPDATE:
                                LOG.trace("Update event. Update");
                                doUpdate(record);
                                break;
                            default:
                                break;
                        }
                    } catch (Exception throwables) {
                        LOG.error("Error preparando la operacion: {}",throwables.getMessage());
                        LOG.error(throwables);
                    }
                    oldOperation = operation;
                    oldSinkTableName = getSourceTableName(record);
                }
            }

            committer.markProcessed(r);
        }

        // Commit transactions
        try {
            if (batchPS != null) {
                int[] rows = batchPS.executeBatch();
                this.getConnection().commit();
                LOG.info("Commited all records. Rows affected: {}", rows.length);
            }
        } catch (SQLException throwables) {
            LOG.error("Un SQLException");
            LOG.error(throwables);
        }catch (Exception e){
            LOG.error("Salgo por aqui");
            LOG.error("Error no controlado: {}", e.getMessage());
            LOG.error(e);
        } finally {
            try {
                if (batchPS != null)
                    batchPS.close();
            } catch (SQLException throwables) {
                LOG.error(throwables);
            }
            batchPS = null;
        }
        committer.markBatchFinished();

    }
Constructs the SqlManager.

@param opts   the ReplicaDB ToolOptions describing the user's requested action.
@param dsType
src/main/java/org/replicadb/manager/cdc/OracleManagerCDC.java