On error when reading transactions from the stream, try reloading the input
stream.
8684
private void tryReloadingEditLog() throws IOException {
    LOG.info("Segment - trying to reload edit log segment");
    // sleep on error
    sleep(errorSleepTimeout);

    // check if a new segment exists
    checkProgress();

    // reopen current segment
    setupIngestStreamWithRetries(currentSegmentTxId);

    // set the position to last good position
    refreshStreamPosition();
  }
645644
private void validate() throws IOException {
        if (null != _error) {
            throw _error;
        }
        
        if(!_isReadLine && !_isReady){
            //If there is no data available then isReady will have returned false and this throw an IllegalStateException
            if (TraceComponent.isAnyTracingEnabled() && tc.isErrorEnabled())
                Tr.error(tc, "read.failed.isReady.false");
            throw new IllegalStateException(Tr.formatMessage(tc, "read.failed.isReady.false"));
        }
    }
1290705
private static Object readDynamic(StreamingInput input)
		throws IOException
	{
		switch(input.peek())
		{
			case VALUE:
				input.next();
				return input.getValue();
			case NULL:
				input.next();
				return input.getValue();
			case LIST_START:
				return readList(input);
			case OBJECT_START:
				return readMap(input);
		}

		throw new SerializationException("Unable to read file, unknown start of value: " + input.peek());
	}
676695
public void resetStream()
        throws IOException, IllegalStateException
    {
        if (isChunking())
            close();
        
        _out=null;
        _nulled=true;
        if (_bufferedOut!=null)
        {
            _bufferedOut.resetStream();
            if (_bufferedOut instanceof ChunkingOutputStream)
                ((ChunkingOutputStream)_bufferedOut).setChunking(false);
        }
        if (_iso8859writer!=null)
            _iso8859writer.flush();
        if (_utf8writer!=null)
            _utf8writer.flush();
        if (_asciiwriter!=null)
            _asciiwriter.flush();

        _bytes=0;
        _written=false;
        _out=_realOut;
        _closing=false;
        _contentLength=-1;
        _nulled=false;
        
        if (_observers!=null)
            _observers.clear();
    }
340062
private synchronized void readObject(java.io.ObjectInputStream stream)
			throws java.io.IOException, ClassNotFoundException
	{
		stream.defaultReadObject();
		this.threadState = (ThreadState) serilizationCache.load(this.threadStateSerilizationHashedId);
		this.threadStateSerilizationHashedId = null;
	}
412640
private final void error() throws ParserException {
	Integer currentState = stateStack.peek();
	parserErrors.addError(currentState);
	if (backtrackEnabled && !backtrackStack.isEmpty()) {
	    trackBack();
	    return;
	}
	if (!backtrackEnabled) {
	    logger.trace("No valid action available and back tracking is disabled. Aborting near '"
		    + getTokenStream().getCodeSample(maxPosition) + "'...");
	} else {
	    logger.trace("No valid action available and back tracking stack is empty. Aborting near '"
		    + getTokenStream().getCodeSample(maxPosition) + "'...");
	}
	throw new ParserException(
		"Error! Could not parse the token stream near '" + getTokenStream().getCodeSample(maxPosition) + "'!");
    }
1012014
public void load(InputStream inputStream)
	{

		try (BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream)))
		{
			while ((currentLine = reader.readLine()) != null)
			{
				lineNumber++;
				currentLine = currentLine.replaceAll("\\s+", " ").trim();
				matcher = linePattern.matcher(currentLine);

				if (matcher.matches())
				{
					String type = matcher.group("type");
					String data = matcher.group("data");

					switch (type)
					{
						case "v":
							addVertex(data);
							break;
						case "vn":
							addNormal(data);
							break;
						case "vt":
							addUV(data);
							break;
						case "f":
							addFace(data);
							break;
						case "g":
						case "o":
							addShape(data);
							break;

						default:
							MalisisCore.log.debug("[ObjFileImporter] Skipped type {} at line {} : {}", type, lineNumber, currentLine);
							break;
					}
				}
				else
				{
					MalisisCore.log.debug("[ObjFileImporter] Skipped non-matching line {} : {}", lineNumber, currentLine);
				}
			}

			addShape("");
		}
		catch (Exception e)
		{
			MalisisCore.log.error("[ObjFileImporter] An error happened while reading the file : {}", e);
		}
	}
8333
private void tryInit() {
    try {
      Configuration conf = getSnapshotConfiguration();
      if (conf != null) {
        // Since this is only used for background loading of transaction snapshots, we use the no-op metrics collector,
        // as there are no relevant metrics to report
        this.storage = new HDFSTransactionStateStorage(conf, new SnapshotCodecProvider(conf),
                                                       new TxMetricsCollector());
        this.storage.startAndWait();
        this.snapshotRefreshFrequency = conf.getLong(TxConstants.Manager.CFG_TX_SNAPSHOT_INTERVAL,
                                                     TxConstants.Manager.DEFAULT_TX_SNAPSHOT_INTERVAL) * 1000;
        this.initialized = true;
      } else {
        LOG.info("Could not load configuration");
      }
    } catch (Exception e) {
      LOG.info("Failed to initialize TransactionStateCache due to: " + e.getMessage());
    }
  }
825584
public void createTransactionFailed(String scope, String streamName) {
        DYNAMIC_LOGGER.incCounterValue(globalMetricName(CREATE_TRANSACTION_FAILED), 1);
        DYNAMIC_LOGGER.incCounterValue(CREATE_TRANSACTION_FAILED, 1, streamTags(scope, streamName));
    }
270313
@Override
    synchronized void restart()
    {
        if (!m_currentReads.isEmpty()) {
            // re-submit all the tasks in the current read set to the pool.
            // the pool will ensure that things submitted with the same
            // txnID will go to the the MpRoSite which is currently running it
            for (TransactionTask task : m_currentReads.values()) {
                taskQueueOffer(task);
            }
        }
        else {
            assert(!m_currentWrites.isEmpty());
            TransactionTask task;
            // There currently should only ever be one current write.  This
            // is the awkward way to get a single value out of a Map
            task = m_currentWrites.entrySet().iterator().next().getValue();
            taskQueueOffer(task);
        }
    }
464245
private void readChunk()
	throws IOException {

	String line = readLine(_in).trim();
	int length  = Integer.parseInt(line, 16);

	if (length > 0) {

	    // make sure the chunk will fit into the buffer

	    if (length > _buf.length) {
		_buf = new byte[length];
	    }

	    int bytesLeft = length;
	    int reqBytes = 0;
	    int off = 0;
	    int read = 0;

	    /*  multiple reads might be necessary to load
		the entire chunk */
	    while (bytesLeft != 0) {
		reqBytes = bytesLeft;
		read = _in.read(_buf, off, reqBytes);
		if (read == -1) break;
		bytesLeft -= read;
		off += read;
	    }

	    _max = off;
	    _index = 0;

	} else {
	    // end of data indicated
	    _eof = true;
	}

	_in.read(); // skip CR
	_in.read(); // skip LF

    }
463978
private void updateStream() throws IOException {
    if (mBlockInStream != null && mBlockInStream.remaining() > 0) { // can still read from stream
      return;
    }

    if (mBlockInStream != null && mBlockInStream.remaining() == 0) { // current stream is done
      closeBlockInStream(mBlockInStream);
    }

    /* Create a new stream to read from mPosition. */
    // Calculate block id.
    long blockId = mStatus.getBlockIds().get(Math.toIntExact(mPosition / mBlockSize));
    // Create stream
    mBlockInStream = mBlockStore.getInStream(blockId, mOptions, mFailedWorkers);
    // Set the stream to the correct position.
    long offset = mPosition % mBlockSize;
    mBlockInStream.seek(offset);
  }
728113
private void openInputStream(File txFile) throws IOException {
    int ioExceptionRetryCount = 0, logHeaderCorruptRetryCount = 0;
    
    LOG.info("Trying to load the edit log from " + txFile.getAbsolutePath());
    
    do {
      try {
        inputStream = new EditLogFileInputStream(txFile);
        editLogFilePosition = inputStream.getPosition();
        curStreamConsumed = false;
        curStreamFinished = false;
        readNullAfterStreamFinished = false;
        LOG.info("Successfully loaded the edits log from " +
            txFile.getAbsolutePath());
        break;
        
      } catch (LogHeaderCorruptException e1) {
        if (logHeaderCorruptRetryCount == LOG_HEADER_CORRUPT_RETRY_MAX) {
          LOG.error("Failed to load the edit log. No retries left.", e1);
          throw new IOException("Could not load the edit log");
        }
        logHeaderCorruptRetryCount ++;
        LOG.warn("Failed to load the edit log. Retry " +
            logHeaderCorruptRetryCount + " ...", e1);
        try {
          Thread.sleep(ioExceptionRetryCount * LOG_HEADER_CORRUPT_BASE_SLEEP);
        } catch (InterruptedException e) {
          throw new IOException(e);
        }
        
      } catch (IOException e2) {
        if (ioExceptionRetryCount == IO_EXCEPTION_RETRY_MAX) {
          LOG.error("Failed to load the edit log. No retries left.", e2);
          throw new IOException("Could not load the edit log");
        }
        ioExceptionRetryCount ++;
        LOG.warn("Failed to load the edit log. Retry " +
            ioExceptionRetryCount + " ...", e2);
        try {
          Thread.sleep(ioExceptionRetryCount * IO_EXCEPTION_BASE_SLEEP);
        } catch (InterruptedException e) {
          throw new IOException(e);
        }
      }
      
    } while (true);
  }
637315
public void reconstitute(BaseDestinationHandler destinationHandler, 
                           SIMPTransactionManager txManager, 
                           ProducerInputHandler inputHandler)
    throws MessageStoreException, SIException
  {
    if (tc.isEntryEnabled())
      SibTr.entry(tc, "reconstitute", new Object[] {destinationHandler, txManager, inputHandler}); 

    super.reconstitute(destinationHandler, txManager);
    
    reconstituteTargetStreams(inputHandler);
          
    if (tc.isEntryEnabled())
      SibTr.exit(tc, "reconstitute");           
  }
149149
protected void attemptFlush() throws SIRollbackException, SIConnectionLostException, SIResourceException, SIErrorException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc,"attemptFlush");

    //The FlushComplete callback
    FlushComplete callback = null;
    //The old set of streams which has just been flushed
    StreamSet oldStreamSet = null;

    // Actions for a flushable stream.  Note that we synchronize
    // up to step 4 to avoid races with threads handling control
    // messages.
    // 1. Cache and unset flushInProgress.
    // 2. Create a new stream ID
    // 3. Remove any data structures associated with the stream.
    // 4. Invoke the callback on the cached FlushComplete reference.
    synchronized (this)
    {
      // Resolve race between control message handlers who both
      // see a flushable stream.
      // flushInProgress was set by startFlush
      if (flushInProgress == null)
      {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(tc, "attemptFlush");
        return;
      }

      // Also verify that the stream really is flushable in case
      // a thread handling a control message called us with a
      // stale source stream reference.
      if (!flushable())
      {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(tc, "attemptFlush");
        return;
      }

      // Now cache the callback.  When we exit the synchronize block,
      // the flush will have officially ended.
      callback = flushInProgress;
      flushInProgress = null;

      // Cleanup the data structures and null out the
      // existing streamSet so we force creation of a new
      // set on the next message.
      oldStreamSet = streamSet;
      streamSet = null;

      if(oldStreamSet != null)
      {
        oldStreamSet.remove();
      }
    }

    // Send out a flushed message.  If this fails, make sure we get
    // to at least invoke the callback.
    try
    {
      // The Cellule argument is null as it is ignored by our parent handler
      // this flush message should be broadcast downstream.
      if (oldStreamSet != null)
        downControl.sendFlushedMessage(null, oldStreamSet.getStreamID());
    }
    catch (Exception e)
    {
      // FFDC
      FFDCFilter.processException(
        e,
        "com.ibm.ws.sib.processor.gd.SourceStreamManager.attemptFlush",
        "1:1158:1.102",
        this);

      // Note that it doesn't make much sense to throw an exception here since
      // this is a callback from stream array map.
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.exit(tc, "attemptFlush", e);
      return;
    }
    finally
    {
      // Invoke the calback in the finally clause so we don't holdup
      // whoever's waiting for the flush to complete.
      callback.flushComplete(destinationHandler);
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "attemptFlush");
  }
138045
public void runInRetryableTransaction(Runnable runnable)
    throws DataAccessException {
        TransactionWrapper transaction=null;
        try {
            transaction = startTransaction();
            runnable.run();
        } catch (Throwable e) {
            if (canRetryTransaction(e)) {
                transaction = (TransactionWrapper)initTransactionRetry(transaction);
                ((ProcessExecutor)getTransactionRetrier()).runInTransaction(runnable);
            }
            else
                throw e;
        } finally {
            stopTransaction(transaction);
        }
    }
1354814
def _txn_retry_loop(conn, callback):
    
    with conn.begin():
        while True:
            try:
                with _NestedTransaction(conn):
                    ret = callback(conn)
                    return ret
            except sqlalchemy.exc.DatabaseError as e:
                if isinstance(e.orig, psycopg2.OperationalError):
                    if e.orig.pgcode == psycopg2.errorcodes.SERIALIZATION_FAILURE:
                        continue
                raise