segmentstore/server/src/main/java/io/pravega/segmentstore/server/logs/MemoryStateUpdater.java
void process(Operation operation) throws DataCorruptionException, CacheFullException {
        // Add entry to MemoryTransactionLog and ReadIndex/Cache. This callback is invoked from the OperationProcessor,
        // which always acks items in order of Sequence Number - so the entries should be ordered (but always check).
        if (operation instanceof StorageOperation) {
            addToReadIndex((StorageOperation) operation);
            if (operation instanceof StreamSegmentAppendOperation) {
                // Transform a StreamSegmentAppendOperation into its corresponding Cached version.
                StreamSegmentAppendOperation appendOp = (StreamSegmentAppendOperation) operation;
                try {
                    operation = new CachedStreamSegmentAppendOperation(appendOp);
                } catch (Throwable ex) {
                    if (Exceptions.mustRethrow(ex)) {
                        throw ex;
                    } else {
                        throw new DataCorruptionException(String.format("Unable to create a CachedStreamSegmentAppendOperation from operation '%s'.", operation), ex);
                    }
                }

                // Release the memory occupied by this StreamSegmentAppendOperation's BufferView - it has been processed
                // and is no longer needed.
                try {
                    appendOp.close();
                } catch (Throwable ex) {
                    if (Exceptions.mustRethrow(ex)) {
                        throw ex;
                    } else {
                        // We do want to know if for some reason we're unable to release the BufferView's memory, but
                        // this is no reason to halt the ingestion pipeline and cause a container shutdown.
                        log.warn("Unable to release memory for operation '{}': ", operation, ex);
                    }
                }
            }
        }

        boolean added = this.inMemoryOperationLog.add(operation);
        if (!added) {
            // This is a pretty nasty one. It's safer to shut down the container than continue.
            // We either recorded the Operation correctly, but invoked this callback out of order, or we really
            // recorded the Operation in the wrong order (by sequence number). In either case, we will be inconsistent
            // while serving reads, so better stop now than later.
            throw new DataCorruptionException("About to have added a Log Operation to InMemoryOperationLog that was out of order.");
        }
    }
/**
     * Processes the given operation and applies it to the ReadIndex and InMemory OperationLog.
     *
     * @param operation The operation to process.
     * @throws DataCorruptionException If a serious, non-recoverable, data corruption was detected, such as trying to
     *                                 append operations out of order.
     * @throws CacheFullException If the operation contains data that needs to be added to the {@link ReadIndex} but it
     * could not be done due to the cache being full and unable to evict anything to make room for more.
     */
Processes the given operation and applies it to the ReadIndex and InMemory OperationLog.
@param operation The operation to process.
@throws DataCorruptionException If a serious, non-recoverable, data corruption was detected, such as trying to
append operations out of order.
@throws CacheFullException If the operation contains data that needs to be added to the  ReadIndex but it
could not be done due to the cache being full and unable to evict anything to make room for more.
Processes the given operation and applies it to the ReadIndex and InMemory OperationLog . @param operation The operation to process . @throws DataCorruptionException If a serious non - recoverable data corruption was detected such as trying to append operations out of order . @throws CacheFullException If the operation contains data that needs to be added to the ReadIndex but it could not be done due to the cache being full and unable to evict anything to make room for more .
void process(Operation operation) throws DataCorruptionException, CacheFullException {
        if (operation instanceof StorageOperation) {
            addToReadIndex((StorageOperation) operation);
            if (operation instanceof StreamSegmentAppendOperation) {
                StreamSegmentAppendOperation appendOp = (StreamSegmentAppendOperation) operation;
                try {
                    operation = new CachedStreamSegmentAppendOperation(appendOp);
                } catch (Throwable ex) {
                    if (Exceptions.mustRethrow(ex)) {
                        throw ex;
                    } else {
                        throw new DataCorruptionException(String.format("Unable to create a CachedStreamSegmentAppendOperation from operation '%s'.", operation), ex);
                    }
                }
                try {
                    appendOp.close();
                } catch (Throwable ex) {
                    if (Exceptions.mustRethrow(ex)) {
                        throw ex;
                    } else {
                        log.warn("Unable to release memory for operation '{}': ", operation, ex);
                    }
                }
            }
        }
        boolean added = this.inMemoryOperationLog.add(operation);
        if (!added) {
            throw new DataCorruptionException("About to have added a Log Operation to InMemoryOperationLog that was out of order.");
        }
    }
void process ( Operation operation ) throws DataCorruptionException , CacheFullException { if ( operation instanceof StorageOperation ) { addToReadIndex ( ( StorageOperation ) operation ) ; if ( operation instanceof StreamSegmentAppendOperation ) { StreamSegmentAppendOperation appendOp = ( StreamSegmentAppendOperation ) operation ; try { operation = new CachedStreamSegmentAppendOperation ( appendOp ) ; } catch ( Throwable ex ) { if ( Exceptions . mustRethrow ( ex ) ) { throw ex ; } else { throw new DataCorruptionException ( String . format ( "Unable to create a CachedStreamSegmentAppendOperation from operation '%s'." , operation ) , ex ) ; } } try { appendOp . close ( ) ; } catch ( Throwable ex ) { if ( Exceptions . mustRethrow ( ex ) ) { throw ex ; } else { log . warn ( "Unable to release memory for operation '{}': " , operation , ex ) ; } } } } boolean added = this . inMemoryOperationLog . add ( operation ) ; if ( ! added ) { throw new DataCorruptionException ( "About to have added a Log Operation to InMemoryOperationLog that was out of order." ) ; } }
Processes the given operation and applies it to the ReadIndex and InMemory OperationLog.
Processes the given operation and applies it to the ReadIndex and InMemory OperationLog .
// Add entry to MemoryTransactionLog and ReadIndex/Cache. This callback is invoked from the OperationProcessor, // which always acks items in order of Sequence Number - so the entries should be ordered (but always check). // Transform a StreamSegmentAppendOperation into its corresponding Cached version. // Release the memory occupied by this StreamSegmentAppendOperation's BufferView - it has been processed // and is no longer needed. // We do want to know if for some reason we're unable to release the BufferView's memory, but // this is no reason to halt the ingestion pipeline and cause a container shutdown. // This is a pretty nasty one. It's safer to shut down the container than continue. // We either recorded the Operation correctly, but invoked this callback out of order, or we really // recorded the Operation in the wrong order (by sequence number). In either case, we will be inconsistent // while serving reads, so better stop now than later.