class BulkLoadProcessor implements AutoCloseable {
    /** Parser of the input bytes. */
    private final BulkLoadParser inputParser;

    /**
     * Converter, which transforms the list of strings parsed from the input stream to the key+value entry to add to
     * the cache.
     */
    private final IgniteClosureX<List<?>, IgniteBiTuple<?, ?>> dataConverter;

    /** Streamer that puts actual key/value into the cache. */
    private final BulkLoadCacheWriter outputStreamer;

    /** Becomes true after {@link #close()} method is called. */
    private boolean isClosed;

    /** Running query manager. */
    private final RunningQueryManager runningQryMgr;

    /** Query id. */
    private final long qryId;

    /** Exception, current load process ended with, or {@code null} if in progress or if succeded. */
    private Exception failReason;

    /** Tracing processor. */
    private final Tracing tracing;

    /** Span of the running query. */
    private final Span qrySpan;

    /**
     * Creates bulk load processor.
     *
     * @param inputParser Parser of the input bytes.
     * @param dataConverter Converter, which transforms the list of strings parsed from the input stream to the
     *     key+value entry to add to the cache.
     * @param outputStreamer Streamer that puts actual key/value into the cache.
     * @param runningQryMgr Running query manager.
     * @param qryId Running query id.
     * @param tracing Tracing processor.
     */
    public BulkLoadProcessor(BulkLoadParser inputParser, IgniteClosureX<List<?>, IgniteBiTuple<?, ?>> dataConverter,
        BulkLoadCacheWriter outputStreamer, RunningQueryManager runningQryMgr, long qryId, Tracing tracing) {
        this.inputParser = inputParser;
        this.dataConverter = dataConverter;
        this.outputStreamer = outputStreamer;
        this.runningQryMgr = runningQryMgr;
        this.qryId = qryId;
        this.tracing = tracing;

        GridRunningQueryInfo qryInfo = runningQryMgr.runningQueryInfo(qryId);

        qrySpan = qryInfo == null ? NoopSpan.INSTANCE : qryInfo.span();

        isClosed = false;
    }

    /**
     * Returns the streamer that puts actual key/value into the cache.
     *
     * @return Streamer that puts actual key/value into the cache.
     */
    public BulkLoadCacheWriter outputStreamer() {
        return outputStreamer;
    }

    /**
     * Processes the incoming batch and writes data to the cache by calling the data converter and output streamer.
     *
     * @param batchData Data from the current batch.
     * @param isLastBatch true if this is the last batch.
     * @throws IgniteIllegalStateException when called after {@link #close()}.
     */
    public void processBatch(byte[] batchData, boolean isLastBatch) throws IgniteCheckedException {<FILL_FUNCTION_BODY>}

    /**
     * Is called to notify processor, that bulk load execution, this processor is performing, failed with specified
     * exception.
     *
     * @param failReason why current load failed.
     */
    public void onError(Exception failReason) {
        this.failReason = failReason;
    }

    /**
     * Aborts processing and closes the underlying objects ({@link IgniteDataStreamer}).
     */
    @Override public void close() throws Exception {
        if (isClosed)
            return;

        try {
            isClosed = true;

            outputStreamer.close();
        }
        catch (Exception e) {
            if (failReason == null)
                failReason = e;

            throw e;
        }
        finally {
            runningQryMgr.unregister(qryId, failReason);
        }
    }
}

        try (TraceSurroundings ignored = MTC.support(tracing.create(SQL_BATCH_PROCESS, qrySpan))) {
            if (isClosed)
                throw new IgniteIllegalStateException("Attempt to process a batch on a closed BulkLoadProcessor");

            Iterable<List<Object>> inputRecords = inputParser.parseBatch(batchData, isLastBatch);

            for (List<Object> record : inputRecords) {
                IgniteBiTuple<?, ?> kv = dataConverter.apply(record);

                outputStreamer.apply(kv);
            }
        }
    
<no_super_class>