Seals the StreamSegment in Storage, if necessary.

@param flushResult The FlushResult from a previous Flush operation. This will just be passed-through.
@param timer       Timer for the operation.
@return The FlushResult passed in as an argument.
426646
private CompletableFuture<WriterFlushResult> sealIfNecessary(WriterFlushResult flushResult, TimeoutTimer timer) {
        if (!this.hasSealPending.get() || !(this.operations.getFirst() instanceof StreamSegmentSealOperation)) {
            // Either no Seal is pending or the next operation is not a seal - we cannot execute a seal.
            return CompletableFuture.completedFuture(flushResult);
        }

        long traceId = LoggerHelpers.traceEnterWithContext(log, this.traceObjectId, "sealIfNecessary");
        CompletableFuture<Void> sealTask;
        if (this.handle.get() == null) {
            // Segment is empty. It might not have been created yet, so don't do anything.
            assert this.metadata.getStorageLength() == 0 : "handle is null but Metadata.StorageLength is non-zero";
            sealTask = CompletableFuture.completedFuture(null);
        } else {
            // Segment is non-empty; it should exist.
            sealTask = this.storage.seal(this.handle.get(), timer.getRemaining());
        }

        return sealTask
                .thenComposeAsync(v -> sealAttributes(timer.getRemaining()), this.executor)
                .handleAsync((v, ex) -> {
                    ex = Exceptions.unwrap(ex);
                    if (ex != null && !(ex instanceof StreamSegmentSealedException)) {
                        // The operation failed, and it was not because the Segment was already Sealed. Throw it again.
                        // We consider the Seal to succeed if the Segment in Storage is already sealed - it's an idempotent operation.
                        if (ex instanceof StreamSegmentNotExistsException) {
                            // The segment does not exist (anymore). This is possible if it got merged into another one.
                            // Begin a reconciliation to determine the actual state.
                            setState(AggregatorState.ReconciliationNeeded);
                        }
                        throw new CompletionException(ex);
                    }

                    updateStatePostSeal();
                    LoggerHelpers.traceLeave(log, this.traceObjectId, "sealIfNecessary", traceId, flushResult);
                    return flushResult;
                }, this.executor);
    }
199155
function gli_stream_flush_file(str) {
    if (str.streaming)
        GlkOte.log('### gli_stream_flush_file called for streaming file!');
    if (!(str.timer_id === null)) {
        clearTimeout(str.timer_id);
    }
    str.timer_id = null;
    Dialog.file_write(str.ref, str.buf);
}
441733
private Future<?> flushDataFrom(List<CommitLogSegment> segments, boolean force)
    {
        if (segments.isEmpty())
            return Futures.immediateFuture(null);
        final ReplayPosition maxReplayPosition = segments.get(segments.size() - 1).getContext();

        // a map of CfId -> forceFlush() to ensure we only queue one flush per cf
        final Map<UUID, ListenableFuture<?>> flushes = new LinkedHashMap<>();

        for (CommitLogSegment segment : segments)
        {
            for (UUID dirtyCFId : segment.getDirtyCFIDs())
            {
                Pair<String,String> pair = Schema.instance.getCF(dirtyCFId);
                if (pair == null)
                {
                    // even though we remove the schema entry before a final flush when dropping a CF,
                    // it's still possible for a writer to race and finish his append after the flush.
                    logger.debug("Marking clean CF {} that doesn't exist anymore", dirtyCFId);
                    segment.markClean(dirtyCFId, segment.getContext());
                }
                else if (!flushes.containsKey(dirtyCFId))
                {
                    String keyspace = pair.left;
                    final ColumnFamilyStore cfs = Keyspace.open(keyspace).getColumnFamilyStore(dirtyCFId);
                    // can safely call forceFlush here as we will only ever block (briefly) for other attempts to flush,
                    // no deadlock possibility since switchLock removal
                    flushes.put(dirtyCFId, force ? cfs.forceFlush() : cfs.forceFlush(maxReplayPosition));
                }
            }
        }

        return Futures.allAsList(flushes.values());
    }
939367
@Override
  protected void flushAndSync(boolean durable) throws IOException {
    if (outputStream == null) {
      throw new IOException("Trying to use aborted output stream!");
    }
    if (doubleBuf.isFlushed()) {
      return;
    }
    // Under the covers, this invokes a method that does not return until
    // we have successfully persisted the buffer to a quorum of bookies.
    doubleBuf.flushTo(outputStream);
  }
899516
public List<PendingEvent> refreshSegmentEventWriters(Consumer<Segment> segmentSealedCallBack) {
        log.info("Refreshing segments for stream {}", stream);
        return updateSegments(Futures.getAndHandleExceptions(
                controller.getCurrentSegments(stream.getScope(), stream.getStreamName()), RuntimeException::new),
                segmentSealedCallBack);
    }
401261
@Override
    public long getStreamSegmentId(String streamSegmentName, boolean updateLastUsed) {
        synchronized (this.lock) {
            StreamSegmentMetadata metadata = this.metadataByName.getOrDefault(streamSegmentName, null);
            if (updateLastUsed && metadata != null) {
                metadata.setLastUsed(getOperationSequenceNumber());
            }

            return metadata != null ? metadata.getId() : NO_STREAM_SEGMENT_ID;
        }
    }
934600
public synchronized void setFlushInterval(long delay, long seconds) {
        if (m_flush != null) {
            m_flush.cancel(false);
            m_flush = null;
        }
        if (seconds > 0) {
            m_flush = m_ses.scheduleAtFixedRate(new Runnable() {
                @Override
                public void run() {
                    try {
                        flush();
                    } catch (Exception e) {
                        loaderLog.error("Failed to flush loader buffer, some tuples may not be inserted.", e);
                    }
                }
            }, delay, seconds, TimeUnit.SECONDS);
        }
    }
445922
public void reportTimer(Timer timer) {
    LOGGER.debug("+++ entering UnsortedWriterReportVisitor.reportTimer +++");

    long count = timer.getCount();
    long timersum = timer.getTimerSum();
    String result = timer.getName() + " timer : count = " + count + ", timerSum = " + timersum + ", mean = " +
      ((double) timersum / count);
    writeStringToWriter(result);
  }
730039
private PollResult peekOrPoll(String subscription, @Nullable Duration claimTtl, int limit) {
        int remaining = limit;
        Map<Coordinate, EventList> rawEvents = ImmutableMap.of();
        Map<Coordinate, Item> uniqueItems = Maps.newHashMap();
        boolean isPeek = claimTtl == null;
        boolean repeatable = !isPeek && claimTtl.toMillis() > 0;
        boolean eventsAvailableForNextPoll = false;
        boolean noMaxPollTimeOut = true;
        int itemsDiscarded = 0;
        Meter eventMeter = isPeek ? _peekedMeter : _polledMeter;

        // Reading raw events from the event store is a significantly faster operation than resolving the events into
        // databus poll events.  This is because the former sequentially loads small event references while the latter
        // requires reading and resolving the effectively random objects associated with those references from the data
        // store.
        //
        // To make the process more efficient this method first polls for "limit" raw events from the event store.
        // Then, up to the first 10 of those raw events are resolved synchronously with a time limit of MAX_POLL_TIME.
        // Any remaining raw events are resolved lazily as the event list is consumed by the caller.  This makes the
        // return time for this method faster and more predictable while supporting polls for more events than can
        // be resolved within MAX_POLL_TIME.  This is especially beneficial for REST clients which may otherwise time
        // out while waiting for "limit" events to be read and resolved.

        Stopwatch stopwatch = Stopwatch.createStarted(_ticker);
        int padding = 0;
        do {
            if (remaining == 0) {
                break;  // Don't need any more events.
            }

            // Query the databus event store.  Consolidate multiple events that refer to the same item.
            ConsolidatingEventSink sink = new ConsolidatingEventSink(remaining + padding);
            boolean more = isPeek ?
                    _eventStore.peek(subscription, sink) :
                    _eventStore.poll(subscription, claimTtl, sink);
            rawEvents = sink.getEvents();

            if (rawEvents.isEmpty()) {
                // No events to be had.
                eventsAvailableForNextPoll = more;
                break;
            }

            // Resolve the raw events in batches of 10 until at least one response item is found for a maximum time of MAX_POLL_TIME.
            do {
                int batchItemsDiscarded = resolvePeekOrPollEvents(subscription, rawEvents, Math.min(10, remaining),
                        (coord, item) -> {
                            // Check whether we've already added this piece of content to the poll result.  If so, consolidate
                            // the two together to reduce the amount of work a client must do.  Note that the previous item
                            // would be from a previous batch of events and it's possible that we have read two different
                            // versions of the same item of content.  This will prefer the most recent.
                            Item previousItem = uniqueItems.get(coord);
                            if (previousItem != null && previousItem.consolidateWith(item)) {
                                _consolidatedMeter.mark();
                            } else {
                                // We have found a new item of content to return!
                                uniqueItems.put(coord, item);
                            }
                        });
                remaining = limit - uniqueItems.size();
                itemsDiscarded += batchItemsDiscarded;
            } while (!rawEvents.isEmpty() && remaining > 0 && stopwatch.elapsed(TimeUnit.MILLISECONDS) < MAX_POLL_TIME.toMillis());

            // There are more events for the next poll if either the event store explicitly said so or if, due to padding,
            // we got more events than "limit", in which case we're likely to unclaim at last one.
            eventsAvailableForNextPoll = more || rawEvents.size() + uniqueItems.size() > limit;
            if (!more) {
                // There are no more events to be had, so exit now
                break;
            }

            // Note: Due to redundant/unknown events, it's possible that the 'events' list is empty even though, if we
            // tried again, we'd find more events.  Try again a few times, but not for more than MAX_POLL_TIME so clients
            // don't timeout the request.  This helps move through large amounts of redundant deltas relatively quickly
            // while also putting a bound on the total amount of work done by a single call to poll().
            padding = 10;
        } while (repeatable && (noMaxPollTimeOut = stopwatch.elapsed(TimeUnit.MILLISECONDS) < MAX_POLL_TIME.toMillis()));

        Iterator<Event> events;
        int approximateSize;
        if (uniqueItems.isEmpty()) {
            // Either there were no raw events or all events found were for redundant or unknown changes.  It's possible
            // that eventually there will be more events, but to prevent a lengthy delay iterating the remaining events
            // quit now and return an empty result.  The caller can always poll again to try to pick up any more events,
            // and if necessary an async drain will kick off a few lines down to assist in clearing the redundant update
            // wasteland.
            events = Iterators.emptyIterator();
            approximateSize = 0;

            // If there are still more unresolved events claimed then unclaim them now
            if (repeatable && !rawEvents.isEmpty()) {
                unclaim(subscription, rawEvents.values());
            }
        } else if (rawEvents.isEmpty()) {
            // All events have been resolved
            events = toEvents(uniqueItems.values()).iterator();
            approximateSize = uniqueItems.size();
            eventMeter.mark(approximateSize);
        } else {
            // Return an event list which contains the first events which were resolved synchronously plus the
            // remaining events from the peek or poll which will be resolved lazily in batches of 25.

            final Map<Coordinate, EventList> deferredRawEvents = Maps.newLinkedHashMap(rawEvents);
            final int initialDeferredLimit = remaining;

            Iterator<Event> deferredEvents = new AbstractIterator<Event>() {
                private Iterator<Event> currentBatch = Iterators.emptyIterator();
                private int remaining = initialDeferredLimit;

                @Override
                protected Event computeNext() {
                    Event next = null;

                    if (currentBatch.hasNext()) {
                        next = currentBatch.next();
                    } else if (!deferredRawEvents.isEmpty() && remaining > 0) {
                        // Resolve the next batch of events
                        try {
                            final List<Item> items = Lists.newArrayList();
                            do {
                                resolvePeekOrPollEvents(subscription, deferredRawEvents, Math.min(remaining, 25),
                                        (coord, item) -> {
                                            // Unlike with the original batch the deferred batch's events are always
                                            // already de-duplicated by coordinate, so there is no need to maintain
                                            // a coordinate-to-item uniqueness map.
                                            items.add(item);
                                        });

                                if (!items.isEmpty()) {
                                    remaining -= items.size();
                                    currentBatch = toEvents(items).iterator();
                                    next = currentBatch.next();
                                }
                            } while (next == null && !deferredRawEvents.isEmpty() && remaining > 0);
                        } catch (Exception e) {
                            // Don't fail; the caller has already received some events.  Just cut the result stream short
                            // now and throw back any remaining events for a future poll.
                            _log.warn("Failed to load additional events during peek/poll for subscription {}", subscription, e);
                        }
                    }

                    if (next != null) {
                        return next;
                    }

                    if (!deferredRawEvents.isEmpty()) {
                        // If we padded the number of raw events it's possible there are more than the caller actually
                        // requested.  Release the extra padded events now.
                        try {
                            unclaim(subscription, deferredRawEvents.values());
                        } catch (Exception e) {
                            // Don't fail, just log a warning.  The claims will eventually time out on their own.
                            _log.warn("Failed to unclaim {} events from subscription {}", deferredRawEvents.size(), subscription, e);
                        }
                    }

                    // Update the metric for the actual number of events returned
                    eventMeter.mark(limit - remaining);

                    return endOfData();
                }
            };

            events = Iterators.concat(toEvents(uniqueItems.values()).iterator(), deferredEvents);
            approximateSize = uniqueItems.size() + deferredRawEvents.size();
        }

        // Try draining the queue asynchronously if there are still more events available and more redundant events were
        // discarded than resolved items found so far.
        // Doing this only in the poll case for now.
        if (repeatable && eventsAvailableForNextPoll && itemsDiscarded > uniqueItems.size()) {
            drainQueueAsync(subscription);
        }

        return new PollResult(events, approximateSize, eventsAvailableForNextPoll);
    }
28624
public synchronized void flush(boolean b) {
        if (com.ibm.websphere.ras.TraceComponent.isAnyTracingEnabled() && LoggingUtil.SESSION_LOGGER_WAS.isLoggable(Level.FINE)) {
            LoggingUtil.SESSION_LOGGER_WAS.entering(methodClassName, methodNames[FLUSH], "(" + b + ") session = " + this);
        }
        cacheLastAccessedTime = b;
        //we're going to call _sessions.put instead of storeSession to handle the case when it hasn't been inserted yet into the backend
        //if it hasn't been inserted we need to call insertSession, otherwise we just call storeSession.
        //_sessions.storeSession(this);
        _sessions.put(this.getId(), this);
        if (com.ibm.websphere.ras.TraceComponent.isAnyTracingEnabled() && LoggingUtil.SESSION_LOGGER_WAS.isLoggable(Level.FINE)) {
            LoggingUtil.SESSION_LOGGER_WAS.exiting(methodClassName, methodNames[FLUSH]);
        }
    }
968932
public Retryer<R> timeout(final TimeLimiter timeLimiter, final long duration, final TimeUnit timeUnit) {
        return withTimelimiter(new AttemptTimelimit<R>() {

            @Override public R call(Callable<R> callable) throws Exception {
                return checkNotNull(timeLimiter, "TimeLimiter cannot be null")
                        .callWithTimeout(callable, duration, checkNotNull(timeUnit), true);
            }
        });
    }
1209577
public static void updateTimer(Optional<Timer> timer, final long duration, final TimeUnit unit) {
    timer.transform(new Function<Timer, Timer>() {
      @Override
      public Timer apply(@Nonnull Timer input) {
        input.update(duration, unit);
        return input;
      }
    });
  }
598732
void recordIterationStarted(AbstractTimer timer) {
        this.iterationId.incrementAndGet();
        this.currentIterationStartTime.set(timer.getElapsed());
        this.lastIterationError.set(false);
    }
1346997
public void checkForTimeouts(long checkTime) {
        TimeSlot nextSlot = this.firstSlot;
        TimerWorkItem entry = null;
        TimeSlot oldSlot = null;

        while (nextSlot != null && checkTime >= nextSlot.timeoutTime) {
            // Timeout all entries here
            int endIndex = nextSlot.lastEntryIndex;
            for (int i = 0; i <= endIndex; i++) {
                // invoke callback if not already cancelled
                entry = nextSlot.entries[i];

                if (entry.state == TimerWorkItem.ENTRY_ACTIVE) {
                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                        Tr.debug(tc, "Found a timeout, calling timerTriggered");
                    }
                    entry.state = TimerWorkItem.ENTRY_CANCELLED;
                    entry.callback.timerTriggered(entry);
                }
            }

            // since we timed out all slot entries, remove it
            oldSlot = nextSlot;
            removeSlot(oldSlot);
            nextSlot = nextSlot.nextEntry;
        }
    }
8937
@Override
    public CompletableFuture<Void> append(String streamSegmentName, byte[] data, Collection<AttributeUpdate> attributeUpdates, Duration timeout) {
        ensureRunning();

        TimeoutTimer timer = new TimeoutTimer(timeout);
        logRequest("append", streamSegmentName, data.length);
        this.metrics.append();
        return this.metadataStore.getOrAssignSegmentId(streamSegmentName, timer.getRemaining(),
                streamSegmentId -> {
                    StreamSegmentAppendOperation operation = new StreamSegmentAppendOperation(streamSegmentId, data, attributeUpdates);
                    return processAttributeUpdaterOperation(operation, timer);
                });
    }
908426
@Nonnull
    private Timer getTimer() {
        return timerReference.updateAndGet(timer -> timer == null ? new Timer(LogConfigurationTask.class.getSimpleName(), true) : timer);
    }
494984
public OneStepDistributedRowLock<K, C> expireLockAfter(long timeout, TimeUnit unit) {
        this.timeout      = timeout;
        this.timeoutUnits = unit;
        return this;
    }