code: @Override
      public void run(final Iterable<Object[]> buffer) throws Exception {
  
          final UpdateableDatastoreConnection con = datastore.openConnection();
          try {
              final Column[] columns = con.getSchemaNavigator().convertToColumns(schemaName, tableName, columnNames);
  
              if (logger.isDebugEnabled()) {
                  logger.debug("Inserting into columns: {}", Arrays.toString(columns));
              }
  
              final UpdateableDataContext dc = con.getUpdateableDataContext();
              dc.executeUpdate(new BatchUpdateScript() {
                  @Override
                  public void run(UpdateCallback callback) {
                      int insertCount = 0;
                      for (Object[] rowData : buffer) {
                          RowInsertionBuilder insertBuilder = callback.insertInto(columns[0].getTable());
                          for (int i = 0; i < columns.length; i++) {
                              insertBuilder = insertBuilder.value(columns[i], rowData[i]);
                          }
  
                          if (logger.isDebugEnabled()) {
                              logger.debug("Inserting: {}", Arrays.toString(rowData));
                          }
  
                          try {
                              insertBuilder.execute();
                              insertCount++;
                              _writtenRowCount.incrementAndGet();
                          } catch (final RuntimeException e) {
                              errorOccurred(rowData, e);
                          }
                      }
  
                      if (insertCount > 0) {
                          _componentContext.publishMessage(new ExecutionLogMessage(insertCount
                                  + " inserts executed"));
                      }
                  }
              });
  
          } finally {
  
              con.close();
          }
      }
docs: Method invoked when flushing the buffer