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 testSnapshotsDisabled() throws Exception {
        MockFaultHandler faultHandler = new MockFaultHandler("SnapshotGenerator");
        MockEmitter emitter = new MockEmitter().setReady();
        AtomicReference<String> disabledReason = new AtomicReference<>();
        try (SnapshotGenerator generator = new SnapshotGenerator.Builder(emitter).
                setFaultHandler(faultHandler).
                setMaxBytesSinceLastSnapshot(1).
                setMaxTimeSinceLastSnapshotNs(0).
                setDisabledReason(disabledReason).
                build()) {
            disabledReason.compareAndSet(null, "we are testing disable()");
            // No snapshots are generated because snapshots are disabled.
            generator.publishLogDelta(TEST_DELTA, TEST_IMAGE, logDeltaManifestBuilder().build());
        }
        assertEquals(Collections.emptyList(), emitter.images());
        faultHandler.maybeRethrowFirstException();
    }