Reading CDC with Java and Go

01-try-cluster.java

try (Cluster cluster = Cluster.builder().addContactPoint(source).build();
     Session session = cluster.connect()) {

02-set-tablename.java

// Create the set of tables we want to watch.
Set<TableName> tables = Collections.singleton(new TableName(keyspace, table));

03-rawchangeconsumer.java

RawChangeConsumer changeConsumer = change -> {
    // Print the change. 
    printChange(change);
    return CompletableFuture.completedFuture(null);
};

04-rawchangeconsumer.java

// Build a shared consumer of changes.
RawChangeConsumer sharedChangeConsumer = change -> {
    // Print the change. 
    printChange(change);
    return CompletableFuture.completedFuture(null);
};

// Build a provider of this shared consumer. 
RawChangeConsumerProvider changeConsumerProvider = threadId -> {
    return sharedChangeConsumer;
};

05-rawchangeconsumerprovider.java

// Build a provider of consumers. 
RawChangeConsumerProvider changeConsumerProvider = threadId -> {
    // Build a consumer of changes.
    RawChangeConsumer changeConsumer = change -> {
        // Print the change. 
        printChange(change);
        return CompletableFuture.completedFuture(null);
    };
    return changeConsumer;
};

06-cdcconsumer.java

// Build a CDCConsumer instance. We start it in a single-thread
// configuration: workersCount(1).
CDCConsumer consumer = CDCConsumerBuilder.builder(session, changeConsumerProvider, tables).workersCount(1).build();

consumer.start();

// Consume changes for 10 seconds.
Thread.sleep(10000);

consumer.stop();

07-private-static-void.java

private static void printChange(RawChange change) {
    ChangeId changeId = change.getId();
    StreamId streamId = changeId.getStreamId();
    ChangeTime changeTime = changeId.getChangeTime();

08-rawchange.java

// Get the operation type, for example: ROW_UPDATE, POST_IMAGE.
RawChange.OperationType operationType = change.getOperationType();

09-changeschema.java

ChangeSchema changeSchema = change.getSchema();

10-list-changeschema.java

List<ChangeSchema.ColumnDefinition> nonCdcColumnDefinitions = changeSchema.getNonCdcColumnDefinitions();

for (ChangeSchema.ColumnDefinition columnDefinition : nonCdcColumnDefinitions) {
    String columnName = columnDefinition.getColumnName();

    // We can get information if this column was a part of primary key
    // in the base table. Note that in CDC log table different columns
    // are part of a primary key (cdc$stream_id, cdc$time, batch_seq_no).
    ChangeSchema.ColumnType baseTableColumnType = columnDefinition.getBaseTableColumnType();

    // Get the information about the data type (as present in CDC log).
    ChangeSchema.DataType logDataType = columnDefinition.getCdcLogDataType();

11-cell-cell.java

Cell cell = change.getCell(columnName);
Object cellValue = cell.getAsObject();

12-integer-intcellvalue.java

Integer intCellValue = cell.getInt();

13-git-clone.sh

git clone https://github.com/scylladb/scylla-cdc-java.git
cd scylla-cdc-java
mvn clean install
cd scylla-cdc-printer
./scylla-cdc-printer -k KEYSPACE -t TABLE -s SOURCE

14-go-get.go

go get https://github.com/scylladb/scylla-cdc-go

15-cluster.go

cluster := gocql.NewCluster(source)
cluster.PoolConfig.HostSelectionPolicy = gocql.TokenAwareHostPolicy(gocql.DCAwareRoundRobinPolicy("local-dc"))
session, err := cluster.CreateSession()
if err != nil {
    log.Fatal(err)
}
defer session.Close()

16-cfg.go

cfg := &scyllacdc.ReaderConfig{
    Session:               session,
    ChangeConsumerFactory: changeConsumerFactory,
    TableNames:            []string{keyspace + "." + table},
    Logger:                log.New(os.Stderr, "", log.Ldate|log.Lmicroseconds|log.Lshortfile),
}

17-func-printerconsumer.go

func printerConsumer(ctx context.Context, tableName string, c scyllacdc.Change) error {
    // Your logic goes here
    return nil
}

var changeConsumerFactory = scyllacdc.MakeChangeConsumerFactoryFromFunc(printerConsumer)

18-type-myconsumer.go

type myConsumer struct { /* ... */ }

func (mc *myConsumer) Consume(ctx context.Context, change scyllacdc.Change) error {
    // Your logic goes here
    return nil
}

func (mc *myConsumer) End() error {
    return nil
}

type myFactory struct { /* ... */ }

func (f *myFactory) CreateChangeConsumer(ctx context.Context, input scyllacdc.CreateChangeConsumerInput) (ChangeConsumer, error)
    return &myConsumer{
        /* ... */
    }, nil
}

19-reader.go

reader, err := scyllacdc.NewReader(context.Background(), cfg)
if err != nil {
    log.Fatal(err)
}

// Stop the consumer after 10 seconds
time.AfterFunc(10*time.Second, func() {
    reader.Stop()
})

if err := reader.Run(context.Background()); err != nil {
    log.Fatal(err)
}

20-fmt.go

fmt.Printf("[%s %s]:\n", c.StreamID, c.Time.String())

21-for.go

for _, ci := range c.Columns() {
    if !strings.HasPrefix(ci.Name, "cdc$") {
        fmt.Println(ci.Name)
    }
}

22-pkraw.go

pkRaw, _ := changeRow.GetValue("pk")
ckRaw, _ := changeRow.GetValue("ck")
v := changeRow.GetAtomicChange("v")

pk := pkRaw.(*int)
ck := ckRaw.(*int)

fmt.Printf("Operation: %s, pk: %s, ck: %s\n", changeRow.GetOperation(),
    nullableIntToStr(pk), nullableIntToStr(ck))

if v.IsDeleted {
    fmt.Printf("  Column v was set to null/deleted\n")
} else {
    vInt := v.Value.(*int)
    if vInt != nil {
        fmt.Printf("  Column v was set to %d\n", *vInt)
    } else {
        fmt.Print("  Column v was not changed\n")
    }
}

23-for.go

for _, r := range c.PreImage {
    fmt.Printf("    %s\n", r)
}
for _, r := range c.Delta {
    fmt.Printf("    %s\n", r)
}
for _, r := range c.PostImage {
    fmt.Printf("    %s\n", r)
}

24-git-clone.sh

git clone https://github.com/scylladb/scylla-cdc-go
cd scylla-cdc-go
go run ./examples/printer/ -keyspace KEYSPACE -table TABLE -source SOURCE