RPC connector for TigerBeetle #3606

batiati · 2025-09-01T14:06:50Z

Implements an input connector for streaming TigerBeetle events via CDC.

The CDC process is stateless, so this connector requires a cache_resource to persist progress and resume from the last acknowledged event when restarted.

Includes documentation and integration tests.
Licensing information in the file header is pending.

For more information about TigerBeetle CDC, see https://docs.tigerbeetle.com/operating/cdc/

Implements an input connector for streaming [TigerBeetle](https://tigerbeetle.com/) events via CDC. The CDC process is stateless, so this connector requires a [`cache_resource`](https://docs.redpanda.com/redpanda-connect/components/caches/about) to persist progress and resume from the last acknowledged event when restarted. - Includes documentation and integration tests. - Licensing information in the file header is pending. For more information about TigerBeetle CDC, see https://docs.tigerbeetle.com/operating/cdc/

CLAassistant · 2025-09-01T14:06:57Z

All committers have signed the CLA.

batiati · 2025-09-01T14:26:58Z

internal/impl/tigerbeetle/input_tigerbeetle.go

+	return service.NewConfigSpec().
+		Beta().
+		Categories("Services").
+		Version("0.0.1").


I suppose this version should be defined only once we know the release, right?

Yeah, we're actually deprecating this method as we update the docs manually these days.

The same for Beta(), when should we remove it?

Yeah you can get rid of that, but it's also harmless to leave in.

batiati · 2025-09-01T14:31:35Z

public/components/cloud/package.go

TODO: Decide whether it will be included in the cloud bundle.

batiati · 2025-09-01T14:37:45Z

internal/impl/tigerbeetle/input_tigerbeetle.go

+	return nil, nil, ctx.Err()
+}
+
+func newTigerbeetleInput(config *service.ParsedConfig, resources *service.Resources) (s service.BatchInput, err error) {


TODO: Decide whether to check the enterprise license.

Suggested change

func newTigerbeetleInput(config *service.ParsedConfig, resources *service.Resources) (s service.BatchInput, err error) {

func newTigerbeetleInput(config *service.ParsedConfig, resources *service.Resources) (s service.BatchInput, err error) {

if err := license.CheckRunningEnterprise(resources); err != nil {

return nil, err

}

Thanks for setting this up. I believe we're happy to go ahead with a FOSS license so no need for license checks 👍

Jeffail

Thanks @batiati, this is great! I left a few very minor comments but only the assert one needs addressing. Extra thanks for adding the integration test ❤️

internal/impl/tigerbeetle/input_tigerbeetle.go

batiati · 2025-09-02T21:04:39Z

Thanks for the review @Jeffail!

The LintRules are really cool! I've implemented the validations and added unit tests for checking valid and invalid configurations.

Also, I replaced assertions that would panic with errors that will cause the connector to shut down.

Also includes unit tests for all custom lint rules.

…n gracefully.

mmatczuk · 2025-09-03T09:52:42Z

internal/impl/tigerbeetle/input_tigerbeetle.go

+}
+
+// Extracts events from TigerBeetle.
+func (input *tigerbeetleInput) produce(ctx context.Context, client tb.Client, timestampLast uint64) error {


nit: how about using short variable name for self?

I guess it would usually be named just i in Go!

I ended up using the TigerStyle coding style, favoring short but meaningful names and avoiding single-letter variables and abbreviations (except for trivial counters/indexes and well-known abbreviations like ctx, config, etc).

Although TigerStyle was originally designed for another programming language (Zig), I think we can apply most of the rules as long as they don't conflict with Go's idiomatic conventions. I don't want this code to feel alien 👽 compared with the rest of the code base!

It's your call, I'd be glad to follow your coding style otherwise.

I had a look at various inputs it feels like the "idiomatic" way in connect would be t.

internal/impl/tigerbeetle/input_tigerbeetle.go

…ct/components/rate_limits/about) to throttle the number of requests made to TigerBeetle.

david-yu · 2025-09-05T20:52:27Z

Hi @batiati Is this PR ready for another re-review?

batiati · 2025-09-05T22:38:52Z

Hi @batiati Is this PR ready for another re-review?

Yes, it is ready for review. I've addressed all comments so far.

batiati · 2025-09-08T11:03:38Z

CI is failing due to CGO_ENABLED: 0.

The TigerBeetle client requires CGO to consume the static library lib_tbclient (the underlying implementation of the VSR client). It includes pre-built libraries in-repo for all supported platforms.

But it seems there's a policy on builds that require external dependencies and CGO, like the ZeroMQ connector:

By default Redpanda Connect does not build with components that require linking to external libraries. If you wish to build Redpanda Connect locally with this component then set the build tag x_benthos_extra:

internal/impl/tigerbeetle/input_tigerbeetle.go

docs/modules/components/pages/inputs/tigerbeetle_cdc.adoc

internal/impl/tigerbeetle/input_tigerbeetle.go

josephwoodward · 2025-09-08T15:37:43Z

@batiati A couple more of minor comments but otherwise it's looking great!

batiati · 2025-09-09T13:43:07Z

@batiati A couple more of minor comments but otherwise it's looking great!

Thanks @josephwoodward! It's ready for another round.

mmatczuk · 2025-09-11T08:39:23Z

Could we add file headers to the Go files?

Jeffail · 2025-09-11T08:53:19Z

CI is failing due to CGO_ENABLED: 0.

The TigerBeetle client requires CGO to consume the static library lib_tbclient (the underlying implementation of the VSR client). It includes pre-built libraries in-repo for all supported platforms.

But it seems there's a policy on builds that require external dependencies and CGO, like the ZeroMQ connector:

By default Redpanda Connect does not build with components that require linking to external libraries. If you wish to build Redpanda Connect locally with this component then set the build tag x_benthos_extra:

Ideally we'd want to find a way of getting pure Go builds so that this isn't hidden away from the majority of users. Would it be possible to use something like https://github.com/ebitengine/purego here?

mmatczuk · 2025-09-11T08:59:19Z

I think we lack batch policy configuration fieldBatchPolicy = "batching"

batiati · 2025-09-11T12:52:38Z

lack batch policy configuration fieldBatchPolicy =

@mmatczuk, initially I went with the batch policy, but some of the properties didn’t make sense for how TigerBeetle streams out CDC events (such as batching by size in bytes). Instead, we use the event_count_max config to limit the maximum number of events produced per batch.

Perhaps the batch policy could still be useful for pipeline composability. Please let me know.

batiati · 2025-09-11T15:05:11Z

Ideally we'd want to find a way of getting pure Go builds so that this isn't hidden away from the majority of users. Would it be possible to use something like https://github.com/ebitengine/purego here?

Unfortunately, Go’s standard library does not expose a general-purpose FFI or dlopen-style interface! That’s how some of the other clients work.

I know CGO can be a huge burden of complexity, but for the TigerBeetle client, adding a third-party dependency would not be worth the tradeoff either.

mmatczuk · 2025-09-11T16:39:50Z

@batiati would you be open to exploring an approach similar to opendal’s Go bindings without CGO.

batiati · 2025-09-11T18:02:37Z

@batiati would you be open to exploring an approach similar to opendal’s Go bindings without CGO.

They also use purego.

https://github.com/apache/opendal/blob/a1b810910c6d288be8bd698521f843c622673c16/bindings/go/go.mod

We are looking for alternatives in the meantime, but for the short term we can't get rid of CGO.

batiati · 2025-09-15T14:16:57Z

but for the short term we can't get rid of CGO.

Hi @Jeffail and @mmatczuk! I did some research to find alternatives, and we really can't remove the CGO dependency anytime soon. However, as we ship the client with pre-built static libraries for multiple targets, we don't depend too heavily on the system's libraries (as is the case with installing libzmq3-dev when the x_benthos_extra flag is present).

Please let me know what you suggest doing to teach the CI to build the TigerBeetle connector conditionally, and which pieces of documentation warnings we should include about it.

Jeffail · 2025-09-19T08:48:38Z

Hey @batiati, we might try and enable CGO for some of our standard distributions of RPCN, but we almost definitely need to exclude this from our fips builds.

In order to proceed I think we should guard all of these files with a CGO build constraint so that non-cgo builds aren't broken. If you could update your PR to do this then we can merge and handle the rest in a follow-up PR, where we'll probably create fips builds with a cherry-picked list of connectors.

batiati commented Sep 1, 2025

View reviewed changes

Jeffail reviewed Sep 2, 2025

View reviewed changes

batiati added 3 commits September 2, 2025 18:22

Add config LintRule.

6cade89

Also includes unit tests for all custom lint rules.

Replace panic with errors that will cause the connector to shut dow…

43f251e

…n gracefully.

s/not defined/not found

38da49f

batiati force-pushed the main branch from 182ba78 to 38da49f Compare September 2, 2025 21:24

mmatczuk reviewed Sep 3, 2025

View reviewed changes

josephwoodward reviewed Sep 4, 2025

View reviewed changes

internal/impl/tigerbeetle/input_tigerbeetle.go Outdated Show resolved Hide resolved

batiati added 2 commits September 5, 2025 09:05

Naming convention: tigerbeetle_cdc

40a7598

Add support for [rate limit](https://docs.redpanda.com/redpanda-conne…

a530427

…ct/components/rate_limits/about) to throttle the number of requests made to TigerBeetle.