protected RecordWriter getRecordWriter(HiveEndPoint endPoint, UserGroupInformation ugi, HiveConf hiveConf) throws StreamingException, IOException, InterruptedException {
    if (ugi == null) {
        return new StrictJsonWriter(endPoint, hiveConf);
    } else {
        try {
            return ugi.doAs((PrivilegedExceptionAction<StrictJsonWriter>) () -> new StrictJsonWriter(endPoint, hiveConf));
        } catch (UndeclaredThrowableException e) {
            Throwable cause = e.getCause();
            if (cause instanceof StreamingException) {
                throw (StreamingException) cause;
            } else {
                throw e;
            }
        }
    }
}
 

protected RecordWriter getRecordWriter(HiveEndPoint endPoint, UserGroupInformation ugi, HiveConf hiveConf) throws StreamingException, IOException, InterruptedException {
    if (ugi == null) {
        return new StrictJsonWriter(endPoint, hiveConf);
    } else {
        return ugi.doAs((PrivilegedExceptionAction<StrictJsonWriter>) () -> new StrictJsonWriter(endPoint, hiveConf));
    }
}
 

protected TransactionBatch nextTxnBatch(final RecordWriter recordWriter)
        throws InterruptedException, TxnBatchFailure {
    LOG.debug("Fetching new Txn Batch for {}", endPoint);
    TransactionBatch batch = null;
    try {
        batch = callWithTimeout(() -> {
            return connection.fetchTransactionBatch(txnsPerBatch, recordWriter); // could block
        });
        batch.beginNextTransaction();
        LOG.debug("Acquired {}. Switching to first txn", batch);
    } catch (TimeoutException | StreamingException e) {
        throw new TxnBatchFailure(endPoint, e);
    }
    return batch;
}
 

protected TransactionBatch nextTxnBatch(final RecordWriter recordWriter)
        throws InterruptedException, TxnBatchFailure {
    LOG.debug("Fetching new Txn Batch for {}", endPoint);
    TransactionBatch batch = null;
    try {
        batch = callWithTimeout(() -> {
            return connection.fetchTransactionBatch(txnsPerBatch, recordWriter); // could block
        });
        batch.beginNextTransaction();
        LOG.debug("Acquired {}. Switching to first txn", batch);
    } catch (TimeoutException | StreamingException e) {
        throw new TxnBatchFailure(endPoint, e);
    }
    return batch;
}
 

@Override
protected RecordWriter getRecordWriter(HiveEndPoint endPoint, UserGroupInformation ugi, HiveConf conf) throws StreamingException {
    assertEquals(hiveConf, conf);
    return mock(RecordWriter.class);
}
 

@Override
protected TransactionBatch nextTxnBatch(RecordWriter recordWriter) throws InterruptedException, TxnBatchFailure {
    TransactionBatch txnBatch = mock(TransactionBatch.class);
    return txnBatch;
}
 

@Override
public RecordWriter createRecordWriter(HiveEndPoint hiveEndPoint) throws StreamingException, IOException, ClassNotFoundException {
  List<String> result = fields.stream().map(String::toLowerCase).collect(Collectors.toList());
  return new DelimitedInputWriter(result.toArray(new String[0]), fieldDelimiter, hiveEndPoint);
}
 

@Override
public RecordWriter load(HiveEndPoint endPoint) throws Exception {
  return new StrictJsonWriter(endPoint, hiveConf);
}
 

@Override
protected RecordWriter getRecordWriter(HiveEndPoint endPoint, UserGroupInformation ugi, HiveConf conf) throws StreamingException {
    assertEquals(hiveConf, conf);
    return mock(RecordWriter.class);
}
 

@Override
protected TransactionBatch nextTxnBatch(RecordWriter recordWriter) throws InterruptedException, TxnBatchFailure {
    TransactionBatch txnBatch = mock(TransactionBatch.class);
    return txnBatch;
}