void publishLogDelta(
        MetadataDelta delta,
        MetadataImage newImage,
        LogDeltaManifest manifest
    ) {
        bytesSinceLastSnapshot += manifest.numBytes();
        if (bytesSinceLastSnapshot >= maxBytesSinceLastSnapshot) {
            if (eventQueue.isEmpty()) {
                scheduleEmit("we have replayed at least " + maxBytesSinceLastSnapshot +
                    " bytes", newImage);
            } else if (log.isTraceEnabled()) {
                log.trace("Not scheduling bytes-based snapshot because event queue is not empty yet.");
            }
        } else if (maxTimeSinceLastSnapshotNs != 0 &&
                (time.nanoseconds() - lastSnapshotTimeNs >= maxTimeSinceLastSnapshotNs)) {
            if (eventQueue.isEmpty()) {
                scheduleEmit("we have waited at least " +
                    TimeUnit.NANOSECONDS.toMinutes(maxTimeSinceLastSnapshotNs) + " minute(s)", newImage);
            } else if (log.isTraceEnabled()) {
                log.trace("Not scheduling time-based snapshot because event queue is not empty yet.");
            }
        } else if (log.isTraceEnabled()) {
            log.trace("Neither time-based nor bytes-based criteria are met; not scheduling snapshot.");
        }
    }
@Test
    public void testTimeBasedSnapshots() throws Exception {
        MockFaultHandler faultHandler = new MockFaultHandler("SnapshotGenerator");
        MockEmitter emitter = new MockEmitter().setReady();
        MockTime mockTime = new MockTime();
        try (SnapshotGenerator generator = new SnapshotGenerator.Builder(emitter).
                setTime(mockTime).
                setFaultHandler(faultHandler).
                setMaxBytesSinceLastSnapshot(200).
                setMaxTimeSinceLastSnapshotNs(TimeUnit.MINUTES.toNanos(30)).
                build()) {
            // This image isn't published yet.
            generator.publishLogDelta(TEST_DELTA, TEST_IMAGE, logDeltaManifestBuilder().numBytes(50).build());
            assertEquals(Collections.emptyList(), emitter.images());
            mockTime.sleep(TimeUnit.MINUTES.toNanos(40));
            // Next image is published because of the time delay.
            generator.publishLogDelta(TEST_DELTA, TEST_IMAGE, logDeltaManifestBuilder().numBytes(50).build());
            TestUtils.waitForCondition(() -> emitter.images().size() == 1, "images.size == 1");
            // bytesSinceLastSnapshot was reset to 0 by the previous snapshot,
            // so this does not trigger a new snapshot.
            generator.publishLogDelta(TEST_DELTA, TEST_IMAGE, logDeltaManifestBuilder().numBytes(150).build());
        }
        assertEquals(Collections.singletonList(TEST_IMAGE), emitter.images());
        faultHandler.maybeRethrowFirstException();
    }