content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.distributedlog;
  
  import com.google.common.annotations.VisibleForTesting;
  import io.netty.buffer.ByteBuf;
  import io.netty.buffer.ByteBufInputStream;
  import io.netty.buffer.Unpooled;
  import io.netty.util.ReferenceCountUtil;
  import java.io.EOFException;
  import java.io.IOException;
  import java.io.InputStream;
  import java.nio.ByteBuffer;
  import javax.annotation.concurrent.NotThreadSafe;
  import org.apache.distributedlog.common.util.ByteBufUtils;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * Log record is the basic element in a log.
   *
   * <p>A log is a sequence of log records. Each log record is a sequence of bytes.
   * Log records are written sequentially into a stream, and will be assigned with
   * an unique system generated sequence number {@link DLSN} (distributedlog sequence
   * number). Besides {@link DLSN}, application can assign its own sequence number
   * while constructing log records. The application defined sequence number is called
   * <code>TransactionID</code> (<i>txid</i>). Either {@link DLSN} or <code>TransactionId</code>
   * could be used to position readers to start from specific log records.
   *
   * <h3>User Record</h3>
   *
   * <p>User records are the records written by applications and read by applications. They
   * are constructed via {@link #LogRecord(long, byte[])} by applications and appended to
   * logs by writers. And they would be deserialized from bytes by the readers and return
   * to applications.
   *
   * <h3>Control Record</h3>
   *
   * <p>Control records are special records that written by distributedlog. They are invisible
   * to applications. They could be treated as <i>commit requests</i> as what people could find
   * in distributed consensus algorithms, since they are usually written by distributedlog to
   * commit application written records. <i>Commit</i> means making application written records
   * visible to readers to achieve consistent views among them.
   *
   * <p>They are named as 'Control Records' for controlling visibility of application written records.
   *
   * <p>The transaction id of 'Control Records' are assigned by distributedlog by inheriting from last
   * written user records. So we could indicate what user records that a control record is committing
   * by looking at its transaction id.
   *
   * <h4>EndOfStream Record</h4>
   *
   * <p><code>EoS</code>(EndOfStream) is a special control record that would be written by a writer
   * to seal a log. After a <i>EoS</i> record is written to a log, no writers could append any record
   * after that and readers will get {@link org.apache.distributedlog.exceptions.EndOfStreamException}
   * when they reach EoS.
   *
   * <p>TransactionID of EoS is <code>Long.MAX_VALUE</code>.
   *
   * <h3>Serialization & Deserialization</h3>
   *
   * <p>Data type in brackets. Interpretation should be on the basis of data types and not individual
   * bytes to honor Endianness.
   *
   * <pre>
   * LogRecord structure:
   * -------------------
   * Bytes 0 - 7                      : Metadata (Long)
   * Bytes 8 - 15                     : TxId (Long)
   * Bytes 16 - 19                    : Payload length (Integer)
   * Bytes 20 - 20+payload.length-1   : Payload (Byte[])
   *
   * Metadata: 8 Bytes (Long)
   * --------
   *
   * 0x 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0
   *            |_____________| |_____|
   *                   |           |
   *               position      flags
   *
   * Flags: 2 Bytes (least significant)
   * -----
   * Bit  0      : If set, control record, else record with payload.
   * Bit  1      : If set, end of stream.
   * Bits 2 - 15 : Unused
   * </pre>
   *
   * <h3>Sequence Numbers</h3>
   *
   * <p>A record is associated with three types of sequence numbers. They are generated
   * and used for different purposes. Check {@link LogRecordWithDLSN} for more details.
   *
   * @see LogRecordWithDLSN
   */
  @NotThreadSafe
  public class LogRecord {
  
      private static final Logger LOG = LoggerFactory.getLogger(LogRecord.class);
  
      // Allow 4K overhead for metadata within the max transmission size
      public static final int MAX_LOGRECORD_SIZE = 1024 * 1024 - 8 * 1024; //1MB - 8KB
      // Allow 4K overhead for transmission overhead
      public static final int MAX_LOGRECORDSET_SIZE = 1024 * 1024 - 4 * 1024; //1MB - 4KB
  
      private static final int INPUTSTREAM_MARK_LIMIT = 16;
  
      static final long LOGRECORD_METADATA_FLAGS_MASK = 0xffffL;
      static final long LOGRECORD_METADATA_FLAGS_UMASK = 0xffffffffffff0000L;
      static final long LOGRECORD_METADATA_POSITION_MASK = 0x0000ffffffff0000L;
      static final long LOGRECORD_METADATA_POSITION_UMASK = 0xffff00000000ffffL;
      static final int LOGRECORD_METADATA_POSITION_SHIFT = 16;
      static final long LOGRECORD_METADATA_UNUSED_MASK = 0xffff000000000000L;
  
      // TODO: Replace with EnumSet
      static final long LOGRECORD_FLAGS_CONTROL_MESSAGE = 0x1;
      static final long LOGRECORD_FLAGS_END_OF_STREAM = 0x2;
      static final long LOGRECORD_FLAGS_RECORD_SET = 0x4;
  
      private long metadata;
      private long txid;
      private ByteBuf payload;
  
      /**
       * Construct an uninitialized log record.
       *
       * <p>NOTE: only deserializer should call this constructor.
       */
      protected LogRecord() {
          this.txid = 0;
          this.metadata = 0;
      }
  
      /**
       * Construct a log record with <i>TransactionId</i> and payload.
       *
       * <p>Usually writer would construct the log record for writing.
       *
       * @param txid
       *          application defined transaction id.
       * @param payload
       *          record data
       */
      public LogRecord(long txid, byte[] payload) {
          this.txid = txid;
          this.payload = Unpooled.wrappedBuffer(payload);
      }
  
      /**
       * Construct a log record with <i>txid</i> and payload <i>buffer</i>.
       *
       * @param txid application defined transaction id.
       * @param buffer payload buffer.
       */
      public LogRecord(long txid, ByteBuffer buffer) {
          this.txid = txid;
          this.payload = Unpooled.wrappedBuffer(buffer);
      }
  
      /**
       * Construct a log record with <i>txid</i> and ByteBuf <i>payload</i>.
       *
       * @param txid transaction id
       * @param payload playload
       */
      public LogRecord(long txid, ByteBuf payload) {
          this.txid = txid;
          this.payload = payload;
          this.metadata = 0;
      }
  
      //
      // Accessors
      //
  
      /**
       * Return application defined transaction id.
       *
       * @return transacton id.
       */
      public long getTransactionId() {
          return txid;
      }
  
      /**
       * Set application defined transaction id.
       *
       * @param txid application defined transaction id.
       */
      protected void setTransactionId(long txid) {
          this.txid = txid;
      }
  
      /**
       * Return the payload of this log record.
       *
       * @return payload of this log record.
       */
      public byte[] getPayload() {
          return ByteBufUtils.getArray(payload);
      }
  
      /**
       * Return the payload buf of this log record.
       *
       * @return payload buf of this record.
       * @since 0.5.0
       */
      public ByteBuf getPayloadBuf() {
          return payload.slice();
      }
  
      void setPayloadBuf(ByteBuf payload, boolean copyData) {
          if (null != this.payload) {
              ReferenceCountUtil.release(this.payload);
          }
          if (copyData) {
              this.payload = Unpooled.copiedBuffer(payload);
          } else {
              this.payload = payload;
          }
      }
  
      /**
       * Return the payload as an {@link InputStream}.
       *
       * @return payload as input stream
       */
      public InputStream getPayLoadInputStream() {
          return new ByteBufInputStream(payload.retainedSlice(), true);
      }
  
      //
      // Metadata & Flags
      //
  
      protected void setMetadata(long metadata) {
          this.metadata = metadata;
      }
  
      protected long getMetadata() {
          return this.metadata;
      }
  
      /**
       * Set the position in the log segment.
       *
       * @see #getPositionWithinLogSegment()
       * @param positionWithinLogSegment position in the log segment.
       */
      void setPositionWithinLogSegment(int positionWithinLogSegment) {
          assert(positionWithinLogSegment >= 0);
          metadata = (metadata & LOGRECORD_METADATA_POSITION_UMASK)
              | (((long) positionWithinLogSegment) << LOGRECORD_METADATA_POSITION_SHIFT);
      }
  
      /**
       * The position in the log segment means how many records (inclusive) added to the log segment so far.
       *
       * @return position of the record in the log segment.
       */
      public int getPositionWithinLogSegment() {
          long ret = (metadata & LOGRECORD_METADATA_POSITION_MASK) >> LOGRECORD_METADATA_POSITION_SHIFT;
          if (ret < 0 || ret > Integer.MAX_VALUE) {
              throw new IllegalArgumentException
                  (ret + " position should never exceed max integer value");
          }
          return (int) ret;
      }
  
      /**
       * Get the last position of this record in the log segment.
       *
       * <p>If the record isn't record set, it would be same as {@link #getPositionWithinLogSegment()},
       * otherwise, it would be {@link #getPositionWithinLogSegment()} + numRecords - 1. If the record set
       * version is unknown, it would be same as {@link #getPositionWithinLogSegment()}.
       *
       * @return last position of this record in the log segment.
       */
      int getLastPositionWithinLogSegment() {
          if (isRecordSet()) {
              try {
                  return getPositionWithinLogSegment() + LogRecordSet.numRecords(this) - 1;
              } catch (IOException e) {
                  // if it is unrecognized record set, we will return the position of this record set.
                  return getPositionWithinLogSegment();
              }
          } else {
              return getPositionWithinLogSegment();
          }
      }
  
      /**
       * Set the record to represent a set of records.
       *
       * <p>The bytes in this record is the serialized format of {@link LogRecordSet}.
       */
      public void setRecordSet() {
          metadata = metadata | LOGRECORD_FLAGS_RECORD_SET;
      }
  
      /**
       * Check if the record represents a set of records.
       *
       * @return true if the record represents a set of records, otherwise false.
       * @see #setRecordSet()
       */
      public boolean isRecordSet() {
          return isRecordSet(metadata);
      }
  
      public static boolean isRecordSet(long metadata) {
          return ((metadata & LOGRECORD_FLAGS_RECORD_SET) != 0);
      }
  
      @VisibleForTesting
      public void setControl() {
          metadata = metadata | LOGRECORD_FLAGS_CONTROL_MESSAGE;
      }
  
      /**
       * Check if the record is a control record.
       *
       * @return true if the record is a control record, otherwise false.
       */
      public boolean isControl() {
          return isControl(metadata);
      }
  
      /**
       * Check flags to see if it indicates a control record.
       *
       * @param flags record flags
       * @return true if the record is a control record, otherwise false.
       */
      public static boolean isControl(long flags) {
          return ((flags & LOGRECORD_FLAGS_CONTROL_MESSAGE) != 0);
      }
  
      /**
       * Set the record as <code>EoS</code> mark.
       *
       * @see #isEndOfStream()
       */
      void setEndOfStream() {
          metadata = metadata | LOGRECORD_FLAGS_END_OF_STREAM;
      }
  
      /**
       * Check if the record is a <code>EoS</code> mark.
       *
       * <p><code>EoS</code> mark is a special record that writer would
       * add to seal a log. after <code>Eos</code> mark is written,
       * writers can't write any more records and readers will get
       * {@link org.apache.distributedlog.exceptions.EndOfStreamException}
       * when they reach <code>EoS</code>.
       *
       * @return true
       */
      boolean isEndOfStream() {
          return ((metadata & LOGRECORD_FLAGS_END_OF_STREAM) != 0);
      }
  
      //
      // Serialization & Deserialization
      //
  
      protected void readPayload(ByteBuf in, boolean copyData) throws IOException {
          int length = in.readInt();
          if (length < 0) {
              throw new EOFException("Log Record is corrupt: Negative length " + length);
          }
          if (copyData) {
              setPayloadBuf(in.slice(in.readerIndex(), length), true);
          } else {
              setPayloadBuf(in.retainedSlice(in.readerIndex(), length), false);
          }
          in.skipBytes(length);
      }
  
      private void writePayload(ByteBuf out) {
          out.writeInt(payload.readableBytes());
          out.writeBytes(payload, payload.readerIndex(), payload.readableBytes());
      }
  
      private void writeToStream(ByteBuf out) {
          out.writeLong(metadata);
          out.writeLong(txid);
          writePayload(out);
      }
  
      /**
       * The size of the serialized log record.
       *
       * <p>This is used to estimate how much will be be appended to the in-memory buffer.
       *
       * @return serialized size
       */
      int getPersistentSize() {
          // Flags + TxId + Payload-length + payload
          return 2 * (Long.SIZE / 8) + Integer.SIZE / 8 + payload.readableBytes();
      }
  
      /**
       * Writer class to write log records into an output {@code stream}.
       */
      public static class Writer {
          private final ByteBuf buf;
  
          public Writer(ByteBuf out) {
              this.buf = out;
          }
  
          /**
           * Write an operation to the output stream.
           *
           * @param record The operation to write
           */
          public void writeOp(LogRecord record) {
              record.writeToStream(buf);
          }
  
          public int getPendingBytes() {
              return buf.readableBytes();
          }
      }
  
      /**
       * Reader class to read log records from an input {@code stream}.
        */
      public static class Reader {
          private final RecordStream recordStream;
          private final ByteBuf in;
          private final long startSequenceId;
          private final boolean deserializeRecordSet;
          private static final int SKIP_BUFFER_SIZE = 512;
          private LogRecordSet.Reader recordSetReader = null;
          private LogRecordWithDLSN lastRecordSkipTo = null;
  
          /**
           * Construct the reader.
           *
           * @param recordStream the record stream for generating {@code DLSN}s.
           * @param in The stream to read from.
           * @param startSequenceId the start sequence id.
           */
          public Reader(RecordStream recordStream,
                        ByteBuf in,
                        long startSequenceId) {
              this(recordStream, in, startSequenceId, true);
          }
  
          public Reader(RecordStream recordStream,
                        ByteBuf in,
                        long startSequenceId,
                        boolean deserializeRecordSet) {
              this.recordStream = recordStream;
              this.in = in;
              this.startSequenceId = startSequenceId;
              this.deserializeRecordSet = deserializeRecordSet;
          }
  
          /**
           * Read an log record from the input stream.
           *
           * <p/> Note that the objects returned from this method may be re-used by future
           * calls to the same method.
           *
           * @return the operation read from the stream, or null at the end of the file
           * @throws IOException on error.
           */
          public LogRecordWithDLSN readOp() throws IOException {
              LogRecordWithDLSN nextRecordInStream;
              while (true) {
                  if (lastRecordSkipTo != null) {
                      nextRecordInStream = lastRecordSkipTo;
                      recordStream.advance(1);
                      lastRecordSkipTo = null;
                      return nextRecordInStream;
                  }
                  if (recordSetReader != null) {
                      nextRecordInStream = recordSetReader.nextRecord();
                      if (null != nextRecordInStream) {
                          recordStream.advance(1);
                          return nextRecordInStream;
                      } else {
                          recordSetReader = null;
                      }
                  }
  
                  if (in.readableBytes() <= 0) {
                      return null;
                  }
  
                  try {
                      long metadata = in.readLong();
                      // Reading the first 8 bytes positions the record stream on the correct log record
                      // By this time all components of the DLSN are valid so this is where we should
                      // retrieve the currentDLSN and advance to the next
                      // Given that there are 20 bytes following the read position of the previous call
                      // to readLong, we should not have moved ahead in the stream.
                      nextRecordInStream = new LogRecordWithDLSN(recordStream.getCurrentPosition(), startSequenceId);
                      nextRecordInStream.setMetadata(metadata);
                      nextRecordInStream.setTransactionId(in.readLong());
  
                      // 1) if it is simple record, copy the data
                      // 2) if it is record set and deserializeRecordSet is true, we don't need to copy data,
                      //    defer data copying to deserializing record from record set.
                      // 3) if it is record set and deserializeRecordSet is false, we copy the data, so applications
                      //    don't have to deal with reference count.
                      boolean copyData = !isRecordSet(metadata) || !deserializeRecordSet;
                      nextRecordInStream.readPayload(in, copyData);
                      if (LOG.isTraceEnabled()) {
                          if (nextRecordInStream.isControl()) {
                              LOG.trace("Reading {} Control DLSN {}",
                                  recordStream.getName(), nextRecordInStream.getDlsn());
                          } else {
                              LOG.trace("Reading {} Valid DLSN {}",
                                  recordStream.getName(), nextRecordInStream.getDlsn());
                          }
                      }
  
                      int numRecords = 1;
                      if (!deserializeRecordSet && nextRecordInStream.isRecordSet()) {
                          numRecords = LogRecordSet.numRecords(nextRecordInStream);
                      }
  
                      if (deserializeRecordSet && nextRecordInStream.isRecordSet()) {
                          recordSetReader = LogRecordSet.of(nextRecordInStream);
                      } else {
                          recordStream.advance(numRecords);
                          return nextRecordInStream;
                      }
                  } catch (EOFException eof) {
                      // Expected
                      break;
                  }
              }
              return null;
          }
  
          public boolean skipTo(long txId, boolean skipControl) throws IOException {
              return skipTo(txId, null, skipControl);
          }
  
          public boolean skipTo(DLSN dlsn) throws IOException {
              return skipTo(null, dlsn, false);
          }
  
          private boolean skipTo(Long txId, DLSN dlsn, boolean skipControl) throws IOException {
              boolean found = false;
              while (true) {
                  try {
                      long flags;
                      long currTxId;
  
                      // if there is not record set, read next record
                      if (null == recordSetReader) {
                          in.markReaderIndex();
                          flags = in.readLong();
                          currTxId = in.readLong();
                      } else {
                          // check record set until reach end of record set
                          lastRecordSkipTo = recordSetReader.nextRecord();
                          if (null == lastRecordSkipTo) {
                              // reach end of record set
                              recordSetReader = null;
                              continue;
                          }
                          flags = lastRecordSkipTo.getMetadata();
                          currTxId = lastRecordSkipTo.getTransactionId();
                      }
  
                      if ((null != dlsn) && (recordStream.getCurrentPosition().compareTo(dlsn) >= 0)) {
                          if (LOG.isTraceEnabled()) {
                              LOG.trace("Found position {} beyond {}", recordStream.getCurrentPosition(), dlsn);
                          }
                          if (null == lastRecordSkipTo) {
                              in.resetReaderIndex();
                          }
                          found = true;
                          break;
                      }
                      if ((null != txId) && (currTxId >= txId)) {
                          if (!skipControl || !isControl(flags)) {
                              if (LOG.isTraceEnabled()) {
                                  LOG.trace("Found position {} beyond {}", currTxId, txId);
                              }
                              if (null == lastRecordSkipTo) {
                                  in.resetReaderIndex();
                              }
                              found = true;
                              break;
                          }
                      }
  
                      if (null != lastRecordSkipTo) {
                          recordStream.advance(1);
                          continue;
                      }
  
                      // get the num of records to skip
                      if (isRecordSet(flags)) {
                          // read record set
                          LogRecordWithDLSN record =
                              new LogRecordWithDLSN(recordStream.getCurrentPosition(), startSequenceId);
                          record.setMetadata(flags);
                          record.setTransactionId(currTxId);
                          record.readPayload(in, false);
                          recordSetReader = LogRecordSet.of(record);
                      } else {
                          int length = in.readInt();
                          if (length < 0) {
                              // We should never really see this as we only write complete entries to
                              // BK and BK client has logic to detect torn writes (through checksum)
                              LOG.info("Encountered Record with negative length at TxId: {}", currTxId);
                              break;
                          }
                          // skip single record
                          in.skipBytes(length);
                          if (LOG.isTraceEnabled()) {
                              LOG.trace("Skipped Record with TxId {} DLSN {}",
                                  currTxId, recordStream.getCurrentPosition());
                          }
                          recordStream.advance(1);
                      }
                  } catch (EOFException eof) {
                      if (LOG.isDebugEnabled()) {
                          LOG.debug("Skip encountered end of file Exception", eof);
                      }
                      break;
                  }
              }
              return found;
          }
      }
  }
  