Repository: diennea/herddb
Pull request #753 (2021-07-20 14:57:23)

Implement first version of Change Data Capture support for HerdDB
Add new utility to tail the HerdDB log from BookKeeper and create a flow
of the Mutations that happen inside a TableSpace.

```
try (ChangeDataCapture cdc = new ChangeDataCapture(
        tableSpaceUUID,
        clientConfiguration,
        new ChangeDataCapture.MutationListener() {
            @Override
            public void accept(ChangeDataCapture.Mutation mutation) {
                LOG.log(Level.INFO, "mutation "+mutation);
                mutations.add(mutation);
            }
        },
        LogSequenceNumber.START_OF_TIME);) {

      cdc.start();

      cdc.run();
 }
```


eolivelli: I would like to add that currently this implementation works if you are able to follow the log from the beginning.
in case you are missing some part of the log we cannot make it work.
We cannot download a snapshot from the current leader because we need to replay all the mutations in the exact order and a snapshot is only a summary with the latest version of the "existing" records
eolivelli: @diegosalvi @aluccaroni @nicoloboschi @amitvc any other comments ?

I have to follow up with another part about storing persistently all the versions of the Table schemas because when you restart the CDC Source you do not have the old versions of each Table.

But I prefer to follow up with another PR
amitvc: LGTM @eolivelli . Can you also add steps on how to use this utility ?
eolivelli: @amitvc currently this is only the first step, you can use the new API to connect to HerdDB and follow the changes on a tablespace.

My plan is to contribute a Pulsar IO Source Connector
https://pulsar.apache.org/docs/en/io-overview/

Patch:
diff --git a/herddb-core/src/main/java/herddb/cdc/ChangeDataCapture.java b/herddb-core/src/main/java/herddb/cdc/ChangeDataCapture.java
new file mode 100644
index 00000000..8b3d72f3
--- /dev/null
+++ b/herddb-core/src/main/java/herddb/cdc/ChangeDataCapture.java
@@ -0,0 +1,318 @@
+/*
+ Licensed to Diennea S.r.l. under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. Diennea S.r.l. licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License.  You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied.  See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+ */
+package herddb.cdc;
+
+
+import static herddb.client.ClientConfiguration.PROPERTY_ZOOKEEPER_ADDRESS;
+import static herddb.client.ClientConfiguration.PROPERTY_ZOOKEEPER_ADDRESS_DEFAULT;
+import static herddb.client.ClientConfiguration.PROPERTY_ZOOKEEPER_PATH;
+import static herddb.client.ClientConfiguration.PROPERTY_ZOOKEEPER_PATH_DEFAULT;
+import static herddb.client.ClientConfiguration.PROPERTY_ZOOKEEPER_SESSIONTIMEOUT;
+import herddb.client.ClientConfiguration;
+import herddb.cluster.BookkeeperCommitLog;
+import herddb.cluster.BookkeeperCommitLogManager;
+import herddb.cluster.ZookeeperMetadataStorageManager;
+import herddb.codec.DataAccessorForFullRecord;
+import herddb.log.CommitLog;
+import herddb.log.LogEntry;
+import herddb.log.LogEntryType;
+import herddb.log.LogSequenceNumber;
+import herddb.metadata.MetadataStorageManagerException;
+import herddb.model.Record;
+import herddb.model.Table;
+import herddb.server.ServerConfiguration;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.bookkeeper.stats.NullStatsLogger;
+
+/**
+ * This utility provides a way to Change Data Capture with HerdDB.
+ */
+public class ChangeDataCapture implements AutoCloseable {
+
+    /**
+     * Type of Mutation
+     */
+    public enum MutationType {
+        INSERT,
+        UPDATE,
+        DELETE,
+        CREATE_TABLE,
+        DROP_TABLE,
+        ALTER_TABLE
+    }
+
+    /**
+     * Details about a Mutation
+     */
+    public static class Mutation {
+        private final Table table;
+        private final MutationType mutationType;
+        private final DataAccessorForFullRecord record;
+        private final LogSequenceNumber logSequenceNumber;
+        private final long timestamp;
+
+        public Mutation(Table table, MutationType mutationType,
+                        DataAccessorForFullRecord record, LogSequenceNumber logSequenceNumber,
+                        long timestamp) {
+            this.table = table;
+            this.mutationType = mutationType;
+            this.record = record;
+            this.logSequenceNumber = logSequenceNumber;
+            this.timestamp = timestamp;
+        }
+
+        public Table getTable() {
+            return table;
+        }
+
+        public MutationType getMutationType() {
+            return mutationType;
+        }
+
+        public DataAccessorForFullRecord getRecord() {
+            return record;
+        }
+
+        public LogSequenceNumber getLogSequenceNumber() {
+            return logSequenceNumber;
+        }
+
+        public long getTimestamp() {
+            return timestamp;
+        }
+
+        @Override
+        public String toString() {
+            return "Mutation{"
+                    + "table=" + table
+                    + ", mutationType=" + mutationType
+                    + ", record=" + record
+                    + ", logSequenceNumber=" + logSequenceNumber
+                    + ", timestamp=" + timestamp
+                    + '}';
+        }
+    }
+
+    /**
+     * Implement this interface in order to receive the flow of Mutations
+     */
+    public interface MutationListener {
+        void accept(Mutation mutation);
+    }
+
+    private final ClientConfiguration configuration;
+    private final MutationListener listener;
+    private LogSequenceNumber lastPosition;
+    private final String tableSpaceUUID;
+    private volatile boolean closed = false;
+    private volatile boolean running = false;
+
+    private ZookeeperMetadataStorageManager zookeeperMetadataStorageManager;
+    private BookkeeperCommitLogManager manager;
+    private Map<String, Table> tablesDefinitions = new HashMap<>();
+    private Map<Long, TransactionHolder> transactions = new HashMap<>();
+
+    private static class TransactionHolder {
+        private List<Mutation> mutations = new ArrayList<>();
+        private Map<String, Table> tablesDefinitions = new HashMap<>();
+    }
+
+    public ChangeDataCapture(String tableSpaceUUID, ClientConfiguration configuration, MutationListener listener, LogSequenceNumber startingPosition) {
+        this.configuration = configuration;
+        this.listener = listener;
+        this.lastPosition = startingPosition;
+        this.tableSpaceUUID = tableSpaceUUID;
+    }
+
+    /**
+     * Bootstrap the procedure.
+     * @throws Exception
+     */
+    public void start() throws Exception {
+        zookeeperMetadataStorageManager = buildMetadataStorageManager(configuration);
+        manager = new BookkeeperCommitLogManager(zookeeperMetadataStorageManager, new ServerConfiguration(), NullStatsLogger.INSTANCE);
+        manager.start();
+    }
+
+    /**
+     * Execute one run
+     * @return the last sequence number, to be used to configure CDC for the next execution
+     * @throws Exception
+     */
+    public LogSequenceNumber run() throws Exception {
+        if (zookeeperMetadataStorageManager == null) {
+            throw new IllegalStateException("not started");
+        }
+
+        try (BookkeeperCommitLog cdc = manager.createCommitLog(tableSpaceUUID, tableSpaceUUID, "cdc");) {
+            running = true;
+            CommitLog.FollowerContext context = cdc.startFollowing(lastPosition);
+            cdc.followTheLeader(lastPosition, new CommitLog.EntryAcceptor() {
+                @Override
+                public boolean accept(LogSequenceNumber lsn, LogEntry entry) throws Exception {
+                    applyEntry(entry, lsn);
+                    lastPosition = lsn;
+                    return !closed;
+                }
+            }, context);
+            return lastPosition;
+        } finally {
+            running = false;
+        }
+    }
+
+    @Override
+    public void close() throws Exception {
+        closed = true;
+        long _start = System.currentTimeMillis();
+
+        while (running
+                && (System.currentTimeMillis() - _start < 10_000)) {
+            Thread.sleep(100);
+        }
+        if (manager != null) {
+            manager.close();
+        }
+        if (zookeeperMetadataStorageManager != null) {
+            zookeeperMetadataStorageManager.close();
+        }
+    }
+
+    private void fire(Mutation mutation, long transactionId) {
+        if (transactionId > 0) {
+            TransactionHolder transaction = transactions.get(transactionId);
+            transaction.mutations.add(mutation);
+        } else {
+            listener.accept(mutation);
+        }
+    }
+
+    private Table lookupTable(LogEntry entry) {
+        String tableName = entry.tableName;
+        if (entry.transactionId > 0) {
+            TransactionHolder transaction = transactions.get(entry.transactionId);
+            Table table = transaction.tablesDefinitions.get(tableName);
+            if (table != null) {
+                return table;
+            }
+        }
+        return tablesDefinitions.get(tableName);
+    }
+
+    private void applyEntry(LogEntry entry, LogSequenceNumber lsn) throws Exception {
+
+        switch (entry.type) {
+            case LogEntryType.NOOP:
+            case LogEntryType.CREATE_INDEX:
+            case LogEntryType.DROP_INDEX:
+                break;
+            case LogEntryType.DROP_TABLE: {
+                Table table = lookupTable(entry);
+
+                if (entry.transactionId > 0) {
+                    TransactionHolder transaction = transactions.get(entry.transactionId);
+                    // set null to mark the table as DROPPED
+                    transaction.tablesDefinitions.put(entry.tableName, null);
+                } else {
+                    tablesDefinitions.remove(entry.tableName, table);
+                }
+                fire(new Mutation(table, MutationType.DROP_TABLE, null, lsn, entry.timestamp), entry.transactionId);
+            }
+            break;
+            case LogEntryType.CREATE_TABLE: {
+                Table table = Table.deserialize(entry.value.to_array());
+                if (entry.transactionId > 0) {
+                    TransactionHolder transaction = transactions.get(entry.transactionId);
+                    transaction.tablesDefinitions.put(entry.tableName, table);
+                } else {
+                    tablesDefinitions.put(entry.tableName, table);
+                }
+                fire(new Mutation(table, MutationType.CREATE_TABLE, null, lsn, entry.timestamp), entry.transactionId);
+            }
+            break;
+            case LogEntryType.ALTER_TABLE: {
+                Table table = Table.deserialize(entry.value.to_array());
+                if (entry.transactionId > 0) {
+                    TransactionHolder transaction = transactions.get(entry.transactionId);
+                    transaction.tablesDefinitions.put(entry.tableName, table);
+                } else {
+                    tablesDefinitions.put(entry.tableName, table);
+                }
+                fire(new Mutation(table, MutationType.ALTER_TABLE, null, lsn, entry.timestamp), entry.transactionId);
+            }
+            break;
+            case LogEntryType.INSERT: {
+                Table table = lookupTable(entry);
+                DataAccessorForFullRecord record = new DataAccessorForFullRecord(table, new Record(entry.key, entry.value));
+                fire(new Mutation(table, MutationType.INSERT, record, lsn, entry.timestamp), entry.transactionId);
+            }
+            break;
+            case LogEntryType.DELETE: {
+                Table table = lookupTable(entry);
+                DataAccessorForFullRecord record = new DataAccessorForFullRecord(table, new Record(entry.key, entry.value));
+                fire(new Mutation(table, MutationType.DELETE, record, lsn, entry.timestamp), entry.transactionId);
+            }
+            break;
+            case LogEntryType.UPDATE: {
+                Table table = lookupTable(entry);
+                DataAccessorForFullRecord record = new DataAccessorForFullRecord(table, new Record(entry.key, entry.value));
+                fire(new Mutation(table, MutationType.UPDATE, record, lsn, entry.timestamp), entry.transactionId);
+            }
+            break;
+            case LogEntryType.BEGINTRANSACTION: {
+                transactions.put(entry.transactionId, new TransactionHolder());
+            }
+            break;
+            case LogEntryType.COMMITTRANSACTION: {
+                TransactionHolder transaction = transactions.remove(entry.transactionId);
+                transaction.tablesDefinitions.forEach((tableName, tableDef) -> {
+                    if (tableDef == null) { // DROP TABLE
+                        tablesDefinitions.remove(tableName);
+                    } else { // CREATE/ALTER
+                        tablesDefinitions.put(tableName, tableDef);
+                    }
+                });
+                for (Mutation mutation : transaction.mutations) {
+                    listener.accept(mutation);
+                }
+            }
+            break;
+            case LogEntryType.ROLLBACKTRANSACTION:
+                transactions.remove(entry.transactionId);
+                break;
+            default:
+                // discard unknown entry types
+                break;
+        }
+    }
+
+    private static ZookeeperMetadataStorageManager buildMetadataStorageManager(ClientConfiguration configuration)
+            throws MetadataStorageManagerException {
+        String zkAddress = configuration.getString(PROPERTY_ZOOKEEPER_ADDRESS, PROPERTY_ZOOKEEPER_ADDRESS_DEFAULT);
+        String zkPath = configuration.getString(PROPERTY_ZOOKEEPER_PATH, PROPERTY_ZOOKEEPER_PATH_DEFAULT);
+        int sessionTimeout = configuration.getInt(PROPERTY_ZOOKEEPER_SESSIONTIMEOUT, 60000);
+        ZookeeperMetadataStorageManager zk = new ZookeeperMetadataStorageManager(zkAddress, sessionTimeout, zkPath);
+        zk.start(false);
+        return zk;
+    }
+}


Test patch:
diff --git a/herddb-core/src/test/java/herddb/cdc/SimpleCDCTest.java b/herddb-core/src/test/java/herddb/cdc/SimpleCDCTest.java
new file mode 100644
index 00000000..5e725671
--- /dev/null
+++ b/herddb-core/src/test/java/herddb/cdc/SimpleCDCTest.java
@@ -0,0 +1,274 @@
+/*
+ Licensed to Diennea S.r.l. under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. Diennea S.r.l. licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License.  You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied.  See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+ */
+
+package herddb.cdc;
+
+import static herddb.core.TestUtils.newServerConfigurationWithAutoPort;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import herddb.client.ClientConfiguration;
+import herddb.codec.RecordSerializer;
+import herddb.core.TestUtils;
+import herddb.log.LogSequenceNumber;
+import herddb.model.Column;
+import herddb.model.ColumnTypes;
+import herddb.model.StatementEvaluationContext;
+import herddb.model.Table;
+import herddb.model.TableSpace;
+import herddb.model.TransactionContext;
+import herddb.model.commands.AlterTableStatement;
+import herddb.model.commands.CreateTableStatement;
+import herddb.model.commands.DeleteStatement;
+import herddb.model.commands.InsertStatement;
+import herddb.model.commands.UpdateStatement;
+import herddb.server.Server;
+import herddb.server.ServerConfiguration;
+import herddb.utils.Bytes;
+import herddb.utils.ZKTestEnv;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+
+/**
+ * Tests around backup/restore
+ *
+ * @author enrico.olivelli
+ */
+public class SimpleCDCTest {
+
+    private static final Logger LOG = Logger.getLogger(SimpleCDCTest.class.getName());
+
+    @Rule
+    public TemporaryFolder folder = new TemporaryFolder();
+
+    private ZKTestEnv testEnv;
+
+    @Before
+    public void beforeSetup() throws Exception {
+        testEnv = new ZKTestEnv(folder.newFolder().toPath());
+        testEnv.startBookieAndInitCluster();
+    }
+
+    @After
+    public void afterTeardown() throws Exception {
+        if (testEnv != null) {
+            testEnv.close();
+        }
+    }
+
+    @Test
+    public void testBasicCaptureDataChange() throws Exception {
+        ServerConfiguration serverconfig_1 = newServerConfigurationWithAutoPort(folder.newFolder().toPath());
+        serverconfig_1.set(ServerConfiguration.PROPERTY_NODEID, "server1");
+        serverconfig_1.set(ServerConfiguration.PROPERTY_MODE, ServerConfiguration.PROPERTY_MODE_CLUSTER);
+        serverconfig_1.set(ServerConfiguration.PROPERTY_ZOOKEEPER_ADDRESS, testEnv.getAddress());
+        serverconfig_1.set(ServerConfiguration.PROPERTY_ZOOKEEPER_PATH, testEnv.getPath());
+        serverconfig_1.set(ServerConfiguration.PROPERTY_ZOOKEEPER_SESSIONTIMEOUT, testEnv.getTimeout());
+
+        ClientConfiguration client_configuration = new ClientConfiguration(folder.newFolder().toPath());
+        client_configuration.set(ClientConfiguration.PROPERTY_MODE, ServerConfiguration.PROPERTY_MODE_CLUSTER);
+        client_configuration.set(ClientConfiguration.PROPERTY_ZOOKEEPER_ADDRESS, testEnv.getAddress());
+        client_configuration.set(ClientConfiguration.PROPERTY_ZOOKEEPER_PATH, testEnv.getPath());
+        client_configuration.set(ClientConfiguration.PROPERTY_ZOOKEEPER_SESSIONTIMEOUT, testEnv.getTimeout());
+
+        try (Server server_1 = new Server(serverconfig_1)) {
+            server_1.start();
+            server_1.waitForStandaloneBoot();
+            Table table = Table.builder()
+                    .name("t1")
+                    .column("c", ColumnTypes.INTEGER)
+                    .column("d", ColumnTypes.INTEGER)
+                    .primaryKey("c")
+                    .build();
+            server_1.getManager().executeStatement(new CreateTableStatement(table), StatementEvaluationContext.DEFAULT_EVALUATION_CONTEXT(), TransactionContext.NO_TRANSACTION);
+            server_1.getManager().executeUpdate(new InsertStatement(TableSpace.DEFAULT, "t1", RecordSerializer.makeRecord(table, "c", 1, "d", 2)), StatementEvaluationContext.DEFAULT_EVALUATION_CONTEXT(), TransactionContext.NO_TRANSACTION);
+            server_1.getManager().executeUpdate(new InsertStatement(TableSpace.DEFAULT, "t1", RecordSerializer.makeRecord(table, "c", 2, "d", 2)), StatementEvaluationContext.DEFAULT_EVALUATION_CONTEXT(), TransactionContext.NO_TRANSACTION);
+            long tx = TestUtils.beginTransaction(server_1.getManager(), TableSpace.DEFAULT);
+
+            server_1.getManager().executeUpdate(new InsertStatement(TableSpace.DEFAULT, "t1", RecordSerializer.makeRecord(table, "c", 3, "d", 2)), StatementEvaluationContext.DEFAULT_EVALUATION_CONTEXT(), new TransactionContext(tx));
+            server_1.getManager().executeUpdate(new InsertStatement(TableSpace.DEFAULT, "t1", RecordSerializer.makeRecord(table, "c", 4, "d", 2)), StatementEvaluationContext.DEFAULT_EVALUATION_CONTEXT(), new TransactionContext(tx));
+
+            List<ChangeDataCapture.Mutation> mutations = new ArrayList<>();
+            try (final ChangeDataCapture cdc = new ChangeDataCapture(
+                    server_1.getManager().getTableSpaceManager(TableSpace.DEFAULT).getTableSpaceUUID(),
+                    client_configuration,
+                    new ChangeDataCapture.MutationListener() {
+                        @Override
+                        public void accept(ChangeDataCapture.Mutation mutation) {
+                            LOG.log(Level.INFO, "mutation " + mutation);
+                            assertTrue(mutation.getTimestamp() > 0);
+                            assertNotNull(mutation.getLogSequenceNumber());
+                            assertNotNull(mutation.getTable());
+                            mutations.add(mutation);
+                        }
+                    },
+                    LogSequenceNumber.START_OF_TIME);) {
+
+                cdc.start();
+
+                cdc.run();
+
+                // we are missing the last entry, because it is not confirmed yet on BookKeeper at this point
+                // also the mutations in the transaction are not visible
+                assertEquals(3, mutations.size());
+
+                // commit the transaction
+                TestUtils.commitTransaction(server_1.getManager(), TableSpace.DEFAULT, tx);
+
+                server_1.getManager().executeUpdate(new UpdateStatement(TableSpace.DEFAULT, "t1", RecordSerializer.makeRecord(table, "c", 4, "d", 2), null), StatementEvaluationContext.DEFAULT_EVALUATION_CONTEXT(), TransactionContext.NO_TRANSACTION);
+
+                cdc.run();
+                assertEquals(5, mutations.size());
+
+                server_1.getManager().executeStatement(new AlterTableStatement(Arrays.asList(Column.column("e", ColumnTypes.INTEGER)), Collections.emptyList(), Collections.emptyList(), null, table.name, TableSpace.DEFAULT, null, Collections.emptyList(),
+                        Collections.emptyList()), StatementEvaluationContext.DEFAULT_EVALUATION_CONTEXT(), TransactionContext.NO_TRANSACTION);
+                cdc.run();
+                assertEquals(6, mutations.size());
+
+
+                // transaction to be rolled back
+                long tx2 = TestUtils.beginTransaction(server_1.getManager(), TableSpace.DEFAULT);
+                server_1.getManager().executeUpdate(new InsertStatement(TableSpace.DEFAULT, "t1", RecordSerializer.makeRecord(table, "c", 30, "d", 2)), StatementEvaluationContext.DEFAULT_EVALUATION_CONTEXT(), new TransactionContext(tx2));
+                server_1.getManager().executeUpdate(new InsertStatement(TableSpace.DEFAULT, "t1", RecordSerializer.makeRecord(table, "c", 31, "d", 2)), StatementEvaluationContext.DEFAULT_EVALUATION_CONTEXT(), new TransactionContext(tx2));
+                TestUtils.roolbackTransaction(server_1.getManager(), TableSpace.DEFAULT, tx2);
+
+                // nothing is to be sent to CDC
+                cdc.run();
+                assertEquals(7, mutations.size());
+
+                server_1.getManager().executeUpdate(new DeleteStatement(TableSpace.DEFAULT, "t1", Bytes.from_int(1), null), StatementEvaluationContext.DEFAULT_EVALUATION_CONTEXT(), TransactionContext.NO_TRANSACTION);
+                cdc.run();
+                assertEquals(7, mutations.size());
+
+                // close the server...close the ledger, now we can read the last mutation
+                server_1.close();
+
+                cdc.run();
+                assertEquals(8, mutations.size());
+
+
+                int i = 0;
+                assertEquals(ChangeDataCapture.MutationType.CREATE_TABLE, mutations.get(i++).getMutationType());
+                assertEquals(ChangeDataCapture.MutationType.INSERT, mutations.get(i++).getMutationType());
+                assertEquals(ChangeDataCapture.MutationType.INSERT, mutations.get(i++).getMutationType());
+                assertEquals(ChangeDataCapture.MutationType.INSERT, mutations.get(i++).getMutationType());
+                assertEquals(ChangeDataCapture.MutationType.INSERT, mutations.get(i++).getMutationType());
+                assertEquals(ChangeDataCapture.MutationType.UPDATE, mutations.get(i++).getMutationType());
+                assertEquals(ChangeDataCapture.MutationType.ALTER_TABLE, mutations.get(i++).getMutationType());
+                assertEquals(ChangeDataCapture.MutationType.DELETE, mutations.get(i++).getMutationType());
+
+
+            }
+        }
+    }
+
+    @Test
+    public void testBasicCaptureDataChangeWithTransactions() throws Exception {
+        ServerConfiguration serverconfig_1 = newServerConfigurationWithAutoPort(folder.newFolder().toPath());
+        serverconfig_1.set(ServerConfiguration.PROPERTY_NODEID, "server1");
+        serverconfig_1.set(ServerConfiguration.PROPERTY_MODE, ServerConfiguration.PROPERTY_MODE_CLUSTER);
+        serverconfig_1.set(ServerConfiguration.PROPERTY_ZOOKEEPER_ADDRESS, testEnv.getAddress());
+        serverconfig_1.set(ServerConfiguration.PROPERTY_ZOOKEEPER_PATH, testEnv.getPath());
+        serverconfig_1.set(ServerConfiguration.PROPERTY_ZOOKEEPER_SESSIONTIMEOUT, testEnv.getTimeout());
+
+        ClientConfiguration client_configuration = new ClientConfiguration(folder.newFolder().toPath());
+        client_configuration.set(ClientConfiguration.PROPERTY_MODE, ServerConfiguration.PROPERTY_MODE_CLUSTER);
+        client_configuration.set(ClientConfiguration.PROPERTY_ZOOKEEPER_ADDRESS, testEnv.getAddress());
+        client_configuration.set(ClientConfiguration.PROPERTY_ZOOKEEPER_PATH, testEnv.getPath());
+        client_configuration.set(ClientConfiguration.PROPERTY_ZOOKEEPER_SESSIONTIMEOUT, testEnv.getTimeout());
+
+        try (Server server_1 = new Server(serverconfig_1)) {
+            server_1.start();
+            server_1.waitForStandaloneBoot();
+            Table table = Table.builder()
+                    .name("t1")
+                    .column("c", ColumnTypes.INTEGER)
+                    .column("d", ColumnTypes.INTEGER)
+                    .primaryKey("c")
+                    .build();
+
+            // create table in transaction
+            long tx = TestUtils.beginTransaction(server_1.getManager(), TableSpace.DEFAULT);
+            server_1.getManager().executeStatement(new CreateTableStatement(table), StatementEvaluationContext.DEFAULT_EVALUATION_CONTEXT(), new TransactionContext(tx));
+            // work on the table in transaction
+            server_1.getManager().executeUpdate(new InsertStatement(TableSpace.DEFAULT, "t1", RecordSerializer.makeRecord(table, "c", 1, "d", 2)), StatementEvaluationContext.DEFAULT_EVALUATION_CONTEXT(), new TransactionContext(tx));
+
+            // commit
+            TestUtils.commitTransaction(server_1.getManager(), TableSpace.DEFAULT, tx);
+
+            // work on the table outside of the transaction
+            server_1.getManager().executeUpdate(new InsertStatement(TableSpace.DEFAULT, "t1", RecordSerializer.makeRecord(table, "c", 2, "d", 2)), StatementEvaluationContext.DEFAULT_EVALUATION_CONTEXT(), TransactionContext.NO_TRANSACTION);
+
+            // close the server and the ledger
+            server_1.close();
+
+            List<ChangeDataCapture.Mutation> mutations = new ArrayList<>();
+            try (final ChangeDataCapture cdc = new ChangeDataCapture(
+                    server_1.getManager().getTableSpaceManager(TableSpace.DEFAULT).getTableSpaceUUID(),
+                    client_configuration,
+                    new ChangeDataCapture.MutationListener() {
+                        @Override
+                        public void accept(ChangeDataCapture.Mutation mutation) {
+                            LOG.log(Level.INFO, "mutation " + mutation);
+                            assertTrue(mutation.getTimestamp() > 0);
+                            assertNotNull(mutation.getLogSequenceNumber());
+                            assertNotNull(mutation.getTable());
+                            mutations.add(mutation);
+                        }
+                    },
+                    LogSequenceNumber.START_OF_TIME);) {
+                cdc.start();
+                cdc.run();
+
+
+                assertEquals(3, mutations.size());
+
+                int i = 0;
+                ChangeDataCapture.Mutation m1 = mutations.get(i++);
+                assertEquals(ChangeDataCapture.MutationType.CREATE_TABLE, m1.getMutationType());
+                Table tableFromM1 = m1.getTable();
+                assertNotNull(tableFromM1);
+                assertEquals(table, tableFromM1);
+                ChangeDataCapture.Mutation m2 = mutations.get(i++);
+                assertEquals(ChangeDataCapture.MutationType.INSERT, m2.getMutationType());
+                assertEquals(m2.getTable(), tableFromM1);
+                assertEquals(1, m2.getRecord().get("c"));
+                assertEquals(2, m2.getRecord().get("d"));
+                ChangeDataCapture.Mutation m3 = mutations.get(i++);
+                assertEquals(ChangeDataCapture.MutationType.INSERT, m3.getMutationType());
+                assertEquals(m3.getTable(), tableFromM1);
+                assertEquals(2, m3.getRecord().get("c"));
+                assertEquals(2, m3.getRecord().get("d"));
+
+            }
+        }
+    }
+
+}
diff --git a/herddb-core/src/test/java/herddb/core/TestUtils.java b/herddb-core/src/test/java/herddb/core/TestUtils.java
index 1431f30d..2b9cd4fd 100644
--- a/herddb-core/src/test/java/herddb/core/TestUtils.java
+++ b/herddb-core/src/test/java/herddb/core/TestUtils.java
@@ -74,6 +74,10 @@ public class TestUtils {
         execute(manager, "COMMIT TRANSACTION '" + tableSpace + "','" + tx + "'", Collections.emptyList(), TransactionContext.NO_TRANSACTION);
     }
 
+    public static void roolbackTransaction(DBManager manager, String tableSpace, long tx) throws StatementExecutionException {
+        execute(manager, "ROLLBACK TRANSACTION '" + tableSpace + "','" + tx + "'", Collections.emptyList(), TransactionContext.NO_TRANSACTION);
+    }
+
     public static StatementExecutionResult execute(DBManager manager, String query, List<Object> parameters) throws StatementExecutionException {
         return execute(manager, query, parameters, TransactionContext.NO_TRANSACTION);
     }