Implementing CDC in Bento via ConduitIO

[nickchomey]

Problem

Using CDC from database change/replication logs has long been an important component in data architectures, and especially for realtime stream processing. Popular tools include Debezium, Kafka Connect, Fivetran and more. None of these are Golang, which makes them cumbersome/expensive.

One of the major changes implemented in Redpanda Connect since the conversion from Benthos was to add a handful of CDC input connectors (mysql, postgres, mongo, gcp spanner), as well as a variety of LLM processors (ollama, openai etc...). It would be great if Bento could get similar capabilities, but unfortunately all of those connectors have an enterprise license, which is, of course, their prerogative.

Proposed Solution

But rather than write your own cdc connectors from scratch, I am proposing a much more seamless and powerful solution: create a new Conduit input/output/processor component that wraps ConduitIO's 60+ CDC source/destination connectors. I have been happily using Conduit, and think it is the ideal tool for CDC-specific tasks, but I sense that their development has come to an end due to business factors. Rather than fork and try to maintain it all in futility, I think that it would be best to just replace the core Conduit pipeline with Bento, which is a very complementary and formidable pairing due to Conduit's focus on CDC and Bento's focus on stream processing.

In fact, I actually previously created a Conduit Processor that was simply a wrapper around an embedded Bento processing pipeline - it received Conduit's record stream, converted them from their (fantastic) OpenCDC format to Bento's service.Message, Bento's pipeline would run and return the transformed records, and then it would convert back to OpenCDC to carry on in the Conduit pipeline. I never actually "used" it, but I'm quite sure it would work well enough (despite surely needing improvements).

What I am proposing here is sort of the opposite: create a Bento input/output component that simply wraps Conduit source/destination connectors so that they can be used in a Bento pipeline. The actual logic for implementing CDC is largely (completely?) independent of Conduit's and Bento's pipeline architecture - the source/input connectors just connect to some data source, take a snapshot, listen for realtime changes, and output them into the pipeline. Destination/output connectors are the same - receive records, connect to a db, and output/write them to it.

A similar thing could probably be done for Conduit's new cohere, openai and ollama processors, so that Bento doesn't need to write its own.

It would be ideal if Bento could support the OpenCDC record format rather than need to convert between that and service.Message - I'd even propose making it the new default for Bento messages, as it makes it simple to interoperate across different connectors.

The primary obstacle with this plan would probably be with regards to Conduit's native support for Avro schemas - schemas are extracted and attached to each record, which can then be used through the processing pipeline and destination connector. But I see that as an opportunity to improve Bento, rather than a roadblock.

Alternatives considered

An alternative would be to simply port the connectors directly into fresh Bento connectors. I suppose this would ultimately be cleaner, but would be much more effort than making 3 Conduit input/output/processor wrapper components (and would also cause more fragmentation
/decoupling from the original connectors)

Let me know what you think!

[gregfurman]

Hey! Thanks for the detailed write up here 😄

I'm not all too familiar with Conduit, or the OpenCDC data format, so I've started to spike out an implementation for this -- mapping conduit's source/destination to bento's input/output.

Will update this discussion thread (and address more of your points) when I've gotten some more context on what this would entail 🍱

[nickchomey]

Great! If you'd like a 2nd set of eyes and some feedback, feel free to share a link to the repo where you're working.

[nickchomey]

Also, I think that in the Bento processor for Conduit that I linked to above, the reason that I needed to translate/convert to and from OpenCDC and service.Message was that OpenCDC stuff was (obviously) not anywhere in Bento's pipeline. If those packages were to be included somewhere/somehow, I think that conversion wouldnt have been needed.

[nickchomey]

Sorry, it might take me another day or two to test this out. Things got in the way unexpectedly.

[gregfurman]

Hey @nickchomey. Here's a proof-of-concept for this on my personal bento fork:

https://github.com/gregfurman/bento/tree/add/conduit/internal/impl/conduit

For backwards compatability, you can define conduit components as you would in the official conduit spec except in a bento pipeline i.e

input:
  conduit:
    plugin: "github.com/conduitio/conduit-connector-generator"
    settings:
      format:
        type: "structured"
        options:
          id: "int"
          name: "string"
          company: "string"
          trial: "bool"
      recordCount: "2"

pipeline:
  processors:
    - conduit:
        plugin: field.set
        settings:
          field: '.Payload.After.test_field'
          value: '10'

output:
  conduit:
    plugin: "github.com/conduitio/conduit-connector-file"
    settings:
      path: ./example.out

This is by no means complete but illustrates how something like this would be done. Couple of considerations:

• I only tested this with builtin plugins.
• When referencing builtin plugins for inputs and outputs, the entire path needs to be supplied i.e plugin: "github.com/conduitio/conduit-connector-generator" for generator

I think given how many extra go dependencies this would end up changing, it'd probably make sense to have it as a standalone distro of Bento instead of being present in the main build.

Would love to hear thoughts!

[nickchomey]

i got bento caches working to store schema stuff, rather than conduit's cache and db mechanism. However, that didnt work for the most important part of CDC: storing the last position that was processed.

It may end up being possible to use bento caches, however they dont have some methods that the conduit db seems to need - eg GetAll keys. So, it'll be best to just stick with with the native cache and persistent db - eg badger. I've moved forward with that approach and its mostly working, but i'm still having trouble with storing/retrieving/using the right position stuff upon restart. I'm sure I can get it working.

However, it makes me uneasy how much of Conduit is now being used - since none of it is seemingly maintained anymore, the idea was to just wrap the connectors and use Bento for the rest. But, it seems difficult to truly do so without so many other mechanisms, mostly related to the position tracking. Use more of the conduit connector sdk, core conduit and related packages - while not instantiating the pipeline, orchestrator etc...

I hope i'll get it all "working" tomorrow and will share a branch for others to review. perhaps you'll find ways to improve it all.

[nickchomey]

I think I have to call it quits on this - I've been spinning my wheels for dozens of hours trying to figure out how to get the connectors to accurately and reliably persist their position state, and to be able to reuse it on restart of bento. I have had parts of it working, via various different approaches, but I cant get the whole thing to work.

I tried just adding on different conduit services, wrappers etc to the base that greg had started. But i found that I was slowly just recreating Conduit.

And then i tried starting from a stripped-down version of conduit's runtime.go file (removing the pipeline, provision, and orchestration services). It mostly worked, but there were discrepancies with position state between snapshot and cdc, and especially across restarts.

I especially think that it is/would be very difficult to get the conduit connectors to be properly acking/nacking through bento.

I really do think that this idea has immense potential, but it would probably require a decent effort from someone more knowledgeable than me about conduit and/or bento. In lieu of that, I think I'll stick with Conduit (and perhaps with my Conduit Processor wrapper of a Bento Processing pipeline) - even though it is unmaintained, it is purpose-built for the cdc, position state etc, whereas Bento is not. Likewise, the connectors are built to fit into that system. I also dont see too much reason for it to suddenly just stop working.

I've dumped an extremely crude, unfinished, branch on github in case anyone wants to look and use them for at least some sort of starting point. https://github.com/nickchomey/bento/commits/conduit_runtime/

The commits are a disaster, and encompass various approaches. It should probably be 3 or 4 branches with many more commits per branch... Take it for whatever its worth

[nickchomey]

Update: I decided to return to this and tidy it up significantly so that others have at least some possibility of collaborating/carrying the torch forward. I'll post a new branch hopefully later today that should at least be able to be run and tested, even if persisting position state across restarts, acking etc isnt fully working.

I just needed to recharge for a couple days.

[gregfurman]

@nickchomey Wow thanks so much for all the effort here. I'll give your CDC implementation a look and see how we can keep this going!

[nickchomey]

by the way @gregfurman, i sent you a message in discord a couple weeks ago about this. Perhaps you didnt see it. Feel free to reply there or here after you've seen it

[nickchomey]

By the way, Conduit was originally conceived as a Golang alternative to Confluent Kafka Connect. Given that Warpstream is now owned by Confluent, and it seems like it's relatively easy to integrate Conduit with Bento, surely there's tremendous synergies to be had here beyond just adding CDC connectors and LLM Processors.

Conduit has a connector that wraps Confluent Kafka Connect Connectors. Though, all the Golang CDC connectors are surely a better pairing for Bento/Conduit.

Likewise, their automatically-extracted schemas use Avro and Confluent Schema Registry, which is what your schema_registry_encode/decode and avro scanner/processor use.

Again, Conduit's CDC + Schema capabilities + Bento's stream processing capabilities are an enormously complementary pairing!

[joeblew999]

Hey @gregfurman

https://github.com/gregfurman/bento/tree/add/conduit/internal/impl/conduit looks like a good start.

I can help with this and getting all the conduit connectors running.

I guess it's not in main branch yet, because it needs dogfooding and more work ?

[nickchomey]

Help would definitely be appreciated with this endeavour, but please see the extensive discussion above.

[qianbeibuzui]

I urgently need the functionality of CDC. How is the progress now? Where can I get progress information?

[gregfurman]

Hey. Which functionality do you need exactly?

@nickchomey has made quite some progess on his fork https://github.com/nickchomey/bento/commits/conduit_runtime/ and I'm looking into how this can be integrated into Bento.

[nickchomey]

As I described in detail above, it would require a very significant engineering effort in order to get this all working properly in bento (unless you want to embed essentially the entire conduit runtime within bento, which seems foolish given that it's unmaintained).

Moreover, I didn't receive any feedback here, so I gave up and am now using debezium, and then just process the messages through bento. Hopefully someday it'll all be possible within a pure go binary. (Redpanda connect has cdc, but it's behind an enterprise license. And also don't look particularly comprehensive)

[qianbeibuzui]

As I described in detail above, it would require a very significant engineering effort in order to get this all working properly in bento (unless you want to embed essentially the entire conduit runtime within bento, which seems foolish given that it's unmaintained).

Moreover, I didn't receive any feedback here, so I gave up and am now using debezium, and then just process the messages through bento. Hopefully someday it'll all be possible within a pure go binary. (Redpanda connect has cdc, but it's behind an enterprise license. And also don't look particularly comprehensive)

My idea is consistent with yours, to use a CDC synchronization tool like Debezium to transmit messages to the message queue and then process them through Bento. I am very eager for Bento to join a tool similar to Flink CDC.

[nickchomey]

Having cdc right in bento/benthos would be ideal. But it's a complicated thing to do well. Debezium is by far the most mature and surely will always be maintained given how many large companies are using it. It's easy to use with Bento - just use a kafka or nats jetstream sink in DBZ and inputin bento. Depending on your needs, you might need a schema registry as well and perhaps some sort of data type schema mapping.