/*
(non-Javadoc)

@see com.ibm.ws.objectManager.LogOutput#writeNext(com.ibm.ws.objectManager.LogRecord, long, boolean)
975015
protected long writeNext(LogRecord logRecord,
                             long reservedDelta,
                             boolean checkSpace,
                             boolean flush)
                    throws ObjectManagerException {
        final String methodName = "writeNext";
        if (Tracing.isAnyTracingEnabled() && trace.isEntryEnabled())
            trace.entry(this,
                        cclass,
                        methodName,
                        new Object[] { logRecord, new Long(reservedDelta), new Boolean(checkSpace), new Boolean(flush) });

        long usableLogSequenceNumber;

        // Take a lock on the log and increment the log sequence number.
        synchronized (logSequenceNumberLock) {
            logSequenceNumber++; // Set the current Sequence number.
            usableLogSequenceNumber = logSequenceNumber;
        } // synchronized (LogSequeunceNumberLock).

        if (Tracing.isAnyTracingEnabled() && trace.isEntryEnabled())
            trace.exit(this,
                       cclass,
                       methodName,
                       new Object[] { new Long(usableLogSequenceNumber) });
        return usableLogSequenceNumber;
    }
113040
BlockCrcInfoWritable getNextRecord() throws IOException {
    // By calling getBucketIdForNextRecord(), we make sure the next field
    // to read is the next record (if there is any record left in the file)
    // Also, by checking the return value, we know whether we've finished
    // the file.
    if (moveToNextRecordAndGetItsBucketId() == -1) {
      return null;
    }

    BlockCrcInfoWritable crcInfo = new BlockCrcInfoWritable();
    crcInfo.readFields(in);
    numRecordsReadInBucket++;
    return crcInfo;
  }
338610
@Override
  public void onNext(final T event) {
    if (LOG.isLoggable(Level.FINE)) {
      LOG.log(Level.FINE, "remoteid: {0}\n{1}", new Object[]{remoteId.getSocketAddress(), event.toString()});
    }
    handler.onNext(new RemoteEvent<T>(myId.getSocketAddress(), remoteId.getSocketAddress(),
        seqGen.getNextSeq(remoteId.getSocketAddress()), event));
  }
41481
@Override
	public void writeStatement(Statement oldStat)
	{
		if (null == oldStat)
		{
			System.err.println("java.lang.Exception: XMLEncoder: discarding statement null");
			System.err.println("Continuing...");
			return;
		}

		// record how the object is changed
		recordStatement(oldStat);
		super.writeStatement(oldStat);
	}
1230919
public EventRecord getNext() {
    try {
	String line = reader.readLine();
	if (line != null) {
	    if (firstLine == null)
		firstLine = new String(line);
	    offset += line.length() + 1;
	    return parseLine(line);
	}
    } catch (IOException e) {
      e.printStackTrace();
    }
    return null;
  }
763057
@Override
  public void write(byte[] record) throws IOException {
    Preconditions.checkNotNull(record);

    byte[] toWrite = record;
    if (this.recordDelimiter.isPresent()) {
      toWrite = Arrays.copyOf(record, record.length + 1);
      toWrite[toWrite.length - 1] = this.recordDelimiter.get();
    }
    if (this.prependSize) {
      long recordSize = toWrite.length;
      ByteBuffer buf = ByteBuffer.allocate(Longs.BYTES);
      buf.putLong(recordSize);
      toWrite = ArrayUtils.addAll(buf.array(), toWrite);
    }
    this.stagingFileOutputStream.write(toWrite);
    this.bytesWritten += toWrite.length;
    this.recordsWritten++;
  }
321467
@Override
    public void putNext(Tuple t) throws IOException {
        pigTuple.setTuple(t);

        if (trace) {
            log.trace("Writing out tuple " + t);
        }
        try {
            writer.write(null, pigTuple);
        } catch (InterruptedException ex) {
            throw new EsHadoopIllegalArgumentException("interrupted", ex);
        }
    }
927511
@Override
	public void writeRecord(Record record) throws IOException {
		int numRecFields = record.getNumFields();
		int readPos;

		for (int i = 0; i < this.numFields; i++) {
			readPos = this.recordPositions[i];
			if (readPos < numRecFields) {
				Value v = record.getField(this.recordPositions[i], this.classes[i]);
				if (v != null) {
					if (i != 0) {
						this.wrt.write(this.fieldDelimiter);
					}
					this.wrt.write(v.toString());
				} else {
					if (this.lenient) {
						if (i != 0) {
							this.wrt.write(this.fieldDelimiter);
						}
					} else {
						throw new RuntimeException("Cannot serialize record with <null> value at position: " + readPos);
					}
				}

			} else {
				if (this.lenient) {
					if (i != 0) {
						this.wrt.write(this.fieldDelimiter);
					}
				} else {
					throw new RuntimeException("Cannot serialize record with out field at position: " + readPos);
				}
			}

		}

		// add the record delimiter
		this.wrt.write(this.recordDelimiter);
	}
452005
void expireAfterWrite(String key, @Nullable String value) {
    requireArgument(expireAfterWriteDuration == UNSET_INT, "expireAfterWrite was already set");
    expireAfterWriteDuration = parseDuration(key, value);
    expireAfterWriteTimeUnit = parseTimeUnit(key, value);
  }
438745
void writeStatement(Session session, String s) {

        try {
            dbLogWriter.writeLogStatement(session, s);
        } catch (IOException e) {
            throw Error.error(ErrorCode.FILE_IO_ERROR, logFileName);
        }

        if (maxLogSize > 0 && dbLogWriter.size() > maxLogSize) {
            database.logger.needsCheckpoint = true;
        }
    }
185206
public void write(DataOutput out) throws IOException {
    taskid.write(out);
    out.writeFloat(progress);
    out.writeInt(numSlots);
    WritableUtils.writeEnum(out, runState);
    Text.writeString(out, diagnosticInfo);
    Text.writeString(out, stateString);
    WritableUtils.writeEnum(out, phase);
    out.writeLong(startTime);
    out.writeLong(finishTime);
    out.writeBoolean(includeCounters);
    out.writeLong(outputSize);
    if (includeCounters) {
      counters.write(out);
    }
    nextRecordRange.write(out);
  }
758563
private java.io.DataInputStream readNextLogRecord()
                    throws ObjectManagerException
    {
        if (Tracing.isAnyTracingEnabled() && trace.isEntryEnabled())
            trace.entry(this,
                        cclass,
                        "readNextLogRecord",
                        new Object[] { new Long(sectorValidatedInputStream.currentPage),
                                      new Integer(sectorValidatedInputStream.currentByte) });

        byte[] logRecordBytes = null;
        short partLength = 0;
        byte partType = FileLogOutput.PART_First;
        byte multiPartID = 0;
        java.io.DataInputStream logRecordDataInputStream = null;

        // Keep reading until we have assembled a complete LogRecord.
        while (logRecordDataInputStream == null) {
            try {
                // Read the part header.
                partType = dataInputStream.readByte();
                switch (partType) {
                    case FileLogOutput.PART_Padding:
                        // Padding, do nothing.
                        break;

                    case FileLogOutput.PART_First:
                    case FileLogOutput.PART_Middle:
                    case FileLogOutput.PART_Last:
                        multiPartID = dataInputStream.readByte();
                        partLength = dataInputStream.readShort();

                        // Now get the logRecord itself.
                        // TODO We could avoid doing this if the logRecord is single part as it is in fact just the next bytes
                        //      in the datastream.
                        logRecordBytes = new byte[partLength];
                        // There is a bug in JVM 1.4.2 such that dataInput.readFully(byte[],int,int)
                        // returns a partial buffer rather than throwing EOFExcepton. 
                        // dataInputStream.readFully(logRecordBytes);
                        // Anyhow we would rather not block if the file has data missing at the end.
                        int bytesRead = dataInputStream.read(logRecordBytes);
                        if (bytesRead != partLength) {
                            if (Tracing.isAnyTracingEnabled() && trace.isEntryEnabled())
                                trace.exit(this, cclass
                                           , "readNextLogRecord"
                                           , new Object[] { "LogFileExhausted_1", new Integer(bytesRead), new Short(partLength) }
                                                );
                            throw new LogFileExhaustedException(this, new java.io.EOFException());
                        } // if ( bytesRead != partLength).

                        if (multiPartID == 0) {
                            // partType will be PART_Last, this is the only part.
                            java.io.ByteArrayInputStream byteArrayInputStream = new java.io.ByteArrayInputStream(logRecordBytes);
                            logRecordDataInputStream = new java.io.DataInputStream(byteArrayInputStream);

                        } else if (partType == FileLogOutput.PART_Last) {
                            if (multiPartLogRecords[multiPartID] != null) {
                                multiPartLogRecords[multiPartID].add(logRecordBytes);
                                MultiByteArrayInputStream inputStream = new MultiByteArrayInputStream(multiPartLogRecords[multiPartID]);
                                logRecordDataInputStream = new java.io.DataInputStream(inputStream);
                                multiPartLogRecords[multiPartID] = null;
                            } // if (multiPartLogRecords[multiPartID] != null).

                        } else if (partType == FileLogOutput.PART_First) {
                            multiPartLogRecords[multiPartID] = new java.util.ArrayList();
                            multiPartLogRecords[multiPartID].add(logRecordBytes);
                        } else {
                            if (multiPartLogRecords[multiPartID] != null) {
                                multiPartLogRecords[multiPartID].add(logRecordBytes);
                            }
                        }

                        break;
                    default:
                        if (Tracing.isAnyTracingEnabled() && trace.isEntryEnabled())
                            trace.exit(this, cclass
                                       , "readNextLogRecord"
                                       , "via InvalidLogPartTypeException partType=" + partType + "(byte)"
                                            );
                        throw new InvalidLogRecordPartTypeException(this
                                                                    , partType);
                        // break;                             // unreachable.
                } // switch (partType).

            } catch (java.io.EOFException exception) {
                // No FFDC Code Needed, condition expected when end of log seen.
                if (Tracing.isAnyTracingEnabled() && trace.isEntryEnabled())
                    trace.exit(this, cclass
                               , "readNextLogRecord"
                               , new Object[] { "LogFileExhausted_2", new Short(partLength) }
                                    );
                throw new LogFileExhaustedException(this, exception);

            } catch (java.io.IOException exception) {
                // No FFDC Code Needed.
                ObjectManager.ffdc.processException(cclass, "readNext", exception, "1:348:1.13");

                if (Tracing.isAnyTracingEnabled() && trace.isEntryEnabled())
                    trace.exit(this, cclass
                               , "readNextLogRecord"
                               , new Object[] { new Short(partLength), exception }
                                    );
                throw new PermanentIOException(this
                                               , exception);
            } // catch java.io.IOException.

        } // while (logRecordDataInputStream == null).

        if (Tracing.isAnyTracingEnabled() && trace.isEntryEnabled())
            trace.exit(this,
                       cclass,
                       "readNextLogRecord",
                       new Object[] { logRecordDataInputStream });
        return logRecordDataInputStream;
    }
709420
@Override
  @SuppressWarnings("checkstyle:illegalcatch")
  public void onNext(final T value) {
    beforeOnNext();
    try {
      executor.submit(new Runnable() {

        @Override
        public void run() {
          try {
            handler.onNext(value);
          } catch (final Throwable t) {
            if (errorHandler != null) {
              errorHandler.onNext(t);
            } else {
              LOG.log(Level.SEVERE, name + " Exception from event handler", t);
              throw t;
            }
          } finally {
            afterOnNext();
          }
        }

      });
    } catch (final Exception e) {
      LOG.log(Level.SEVERE, "Encountered error when submitting to executor in ThreadPoolStage.");
      afterOnNext();
      throw e;
    }

  }
1213253
private void performNextWrite() {
        Append append = getNextAppend();
        if (append == null) {
            return;
        }
        long traceId = LoggerHelpers.traceEnter(log, "storeAppend", append);
        Timer timer = new Timer();
        storeAppend(append)
                .whenComplete((v, e) -> {
                    handleAppendResult(append, e, timer);
                    LoggerHelpers.traceLeave(log, "storeAppend", traceId, v, e);
                })
                .whenComplete((v, e) -> append.getData().release());
    }
495041
@Override
	public BasicLogRecord next() {
		if (!isForward) {
			currentRec = currentRec - pointerSize;
			isForward = true;
		}
		if (currentRec == 0)
			moveToNextBlock();
		currentRec = (Integer) pg.getVal(currentRec, INTEGER).asJavaVal();
		return new BasicLogRecord(pg, new LogSeqNum(blk.number(), currentRec + pointerSize * 2));
	}
423982
private void writeObject(ObjectOutputStream out) throws IOException {

        final boolean isTraceOn = TraceComponent.isAnyTracingEnabled();
        if (isTraceOn && tc.isEntryEnabled())
            Tr.entry(tc, "writeObject: " + this);

        // Use v1 unless features are present that require v2.
        int version = Constants.TIMER_TASK_V1;
        if (parsedSchedule != null) {
            version = Constants.TIMER_TASK_V2;
        }

        out.defaultWriteObject();

        // Write out header information first.
        out.write(EYECATCHER);
        out.writeShort(PLATFORM);
        out.writeShort(version);

        // Write out the instance data.
        out.writeObject(j2eeName.getBytes());
        out.writeObject(userInfoBytes);

        switch (version) {

            case Constants.TIMER_TASK_V1:
                out.writeLong(expiration);
                out.writeLong(interval);
                break;

            case Constants.TIMER_TASK_V2:
                out.writeObject(parsedSchedule);
                out.writeInt(methodId);
                out.writeObject(automaticMethodName);
                out.writeObject(automaticClassName);
                break;

            default:
                // cannot occur since initialize above
                break;
        }

        if (isTraceOn && tc.isEntryEnabled())
            Tr.exit(tc, "writeObject");
    }
257537
@Override
  public synchronized void onNext(final RuntimeStart startTime) {
    if (this.isPidNotWritten()) {
      final long pid = OSUtils.getPID();
      final File outfile = new File(PID_FILE_NAME);
      LOG.log(Level.FINEST, "Storing pid `" + pid + "` in file " + outfile.getAbsolutePath());
      try (final PrintWriter p = new PrintWriter(PID_FILE_NAME, "UTF-8")) {
        p.write(String.valueOf(pid));
        p.write("\n");
      } catch (final FileNotFoundException | UnsupportedEncodingException e) {
        LOG.log(Level.WARNING, "Unable to create PID file.", e);
      }
      this.pidIsWritten = true;
    } else {
      LOG.log(Level.FINEST, "PID already written.");
    }
  }