    @Test
    public void testSnapshotEmptyTable() throws Exception {

        CassandraConnectorContext context = generateTaskContext();
        AtomicBoolean globalTaskState = new AtomicBoolean(true);
        SnapshotProcessor snapshotProcessor = Mockito.spy(new SnapshotProcessor(context));
        when(snapshotProcessor.isRunning()).thenReturn(true);

        context.getCassandraClient().execute("CREATE TABLE IF NOT EXISTS " + keyspaceTable("cdc_table") + " (a int, b text, PRIMARY KEY(a)) WITH cdc = true;");
        context.getSchemaHolder().refreshSchemas();

        ChangeEventQueue<Event> queue = context.getQueue();
        assertEquals(queue.totalCapacity(), queue.remainingCapacity());
        snapshotProcessor.process(); // records empty table to snapshot.offset, so it won't be snapshotted again
        assertEquals(queue.totalCapacity(), queue.remainingCapacity());

        int tableSize = 5;
        for (int i = 0; i < tableSize; i++) {
            context.getCassandraClient().execute("INSERT INTO " + keyspaceTable("cdc_table") + "(a, b) VALUES (?, ?)", i, String.valueOf(i));
        }
        snapshotProcessor.process();
        assertEquals(queue.totalCapacity(), queue.remainingCapacity()); // newly inserted records should be processed by commit log processor instead

        deleteTestKeyspaceTables();
        deleteTestOffsets(context);
        globalTaskState.set(false);
        context.cleanUp();
    }