@Override
    public void add(SegmentOperation operation) throws DataCorruptionException {
        Exceptions.checkNotClosed(this.closed.get(), this);
        Preconditions.checkArgument(operation.getStreamSegmentId() == this.connector.getMetadata().getId(),
                "Operation '%s' refers to a different Segment than this one (%s).", operation, this.connector.getMetadata().getId());
        Preconditions.checkArgument(operation.getSequenceNumber() != Operation.NO_SEQUENCE_NUMBER,
                "Operation '%s' does not have a Sequence Number assigned.", operation);
        if (this.connector.getMetadata().isDeleted() || !(operation instanceof CachedStreamSegmentAppendOperation)) {
            // Segment is either deleted or this is not an append operation. Nothing for us to do here.
            return;
        }

        CachedStreamSegmentAppendOperation append = (CachedStreamSegmentAppendOperation) operation;
        if (this.lastAddedOffset.get() >= 0) {
            // We have processed at least one operation so far. Verify operations are contiguous.
            if (this.lastAddedOffset.get() != append.getStreamSegmentOffset()) {
                throw new DataCorruptionException(String.format("Wrong offset for Operation '%s'. Expected: %s, actual: %d.",
                        operation, this.lastAddedOffset, append.getStreamSegmentOffset()));
            }
        } else {
            // We haven't processed any operation so far (this is the first). Verify that we are resuming from an expected
            // offset and not skipping any updates.
            if (this.aggregator.getLastIndexedOffset() < append.getStreamSegmentOffset()) {
                throw new DataCorruptionException(String.format("Operation '%s' begins after TABLE_INDEXED_OFFSET. Expected: %s, actual: %d.",
                        operation, this.aggregator.getLastIndexedOffset(), append.getStreamSegmentOffset()));
            }
        }

        if (append.getStreamSegmentOffset() >= this.aggregator.getLastIndexedOffset()) {
            // Operation has not been indexed yet; add it to the internal list so we can process it.
            // NOTE: appends that contain more than one TableEntry (for batch updates) will be indexed atomically (either
            // all Table Entries are indexed or none), so it is safe to compare this with the first offset of the append.
            this.aggregator.add(append);
            this.lastAddedOffset.set(append.getLastStreamSegmentOffset());
            log.debug("{}: Add {} (State={}).", this.traceObjectId, operation, this.aggregator);
        } else {
            log.debug("{}: Skipped {} (State={}).", this.traceObjectId, operation, this.aggregator);
        }
    }
@ Override public void add ( SegmentOperation operation ) throws DataCorruptionException { Exceptions . checkNotClosed ( this . closed . get ( ) , this ) ; Preconditions . checkArgument ( operation . getStreamSegmentId ( ) == this . connector . getMetadata ( ) . getId ( ) , "Operation '%s' refers to a different Segment than this one (%s)." , operation , this . connector . getMetadata ( ) . getId ( ) ) ; Preconditions . checkArgument ( operation . getSequenceNumber ( ) != Operation . NO_SEQUENCE_NUMBER , "Operation '%s' does not have a Sequence Number assigned." , operation ) ; if ( this . connector . getMetadata ( ) . isDeleted ( ) || ! ( operation instanceof CachedStreamSegmentAppendOperation ) ) { return ; } CachedStreamSegmentAppendOperation append = ( CachedStreamSegmentAppendOperation ) operation ; if ( this . lastAddedOffset . get ( ) >= 0 ) { if ( this . lastAddedOffset . get ( ) != append . getStreamSegmentOffset ( ) ) { throw new DataCorruptionException ( String . format ( "Wrong offset for Operation '%s'. Expected: %s, actual: %d." , operation , this . lastAddedOffset , append . getStreamSegmentOffset ( ) ) ) ; } } else { if ( this . aggregator . getLastIndexedOffset ( ) < append . getStreamSegmentOffset ( ) ) { throw new DataCorruptionException ( String . format ( "Operation '%s' begins after TABLE_INDEXED_OFFSET. Expected: %s, actual: %d." , operation , this . aggregator . getLastIndexedOffset ( ) , append . getStreamSegmentOffset ( ) ) ) ; } } if ( append . getStreamSegmentOffset ( ) >= this . aggregator . getLastIndexedOffset ( ) ) { this . aggregator . add ( append ) ; this . lastAddedOffset . set ( append . getLastStreamSegmentOffset ( ) ) ; log . debug ( "{}: Add {} (State={})." , this . traceObjectId , operation , this . aggregator ) ; } else { log . debug ( "{}: Skipped {} (State={})." , this . traceObjectId , operation , this . aggregator ) ; } }
region WriterSegmentProcessor Implementation