RollingCountBolt.java

public void execute(Tuple input) {
    if (isTickTuple(input)) {
        LOGGER.debug("Received tick tuple, triggering emit of current window counts");
        emitCurrentWindowCounts();
    } else {
        countObj(input);
    }
    collector.ack(input);
}

private void emitCurrentWindowCounts() {
    Map<Object, Long> counts = counter.getCountsThenAdvanceWindow();
    int actualWindowLengthInSeconds = lastModifiedTracker.secondsSinceOldestModification();
    lastModifiedTracker.markAsModified();
    if (actualWindowLengthInSeconds < windowLengthInSeconds-1) {
        LOGGER.debug(String.format(WINDOW_LENGTH_WARNING_TEMPLATE, actualWindowLengthInSeconds, windowLengthInSeconds));
    }
    emit(counts, actualWindowLengthInSeconds);
}

private void emit(Map<Object, Long> counts, int actualWindowLengthInSeconds) {
    for (Entry<Object, Long> entry : counts.entrySet()) {
        Object obj = entry.getKey();
        Long count = entry.getValue();
        collector.emit(new Values(obj, count, actualWindowLengthInSeconds));
    }
}