kafka: adding best effort and consistent mode.

[sravotto]

This change adds transactional support to the Kafka connector for changefeeds.

This change uses the conveyor package to deliver mutation to the target database in any of the supported modes of operations.

The conveyor tracks resolved timestamps across all the partitions with a topic, and ensures that the checkpoint used by cdc-sink is advanced based on the minimal resolved timestamp received on all the partitions.

A second commit, refactors the configuration for better readability.

---

This change is 

[codecov-commenter]

Codecov Report

Attention: Patch coverage is 76.34409% with 22 lines in your changes are missing coverage. Please review.

Project coverage is 75.78%. Comparing base (4dcfcdd) to head (c6c4efb).

Files	Patch %	Lines
internal/source/kafka/consumer.go	68.62%	10 Missing and 6 partials ⚠️
internal/source/kafka/conn.go	57.14%	3 Missing ⚠️
internal/source/kafka/config.go	91.30%	1 Missing and 1 partial ⚠️
internal/source/kafka/provider.go	90.90%	0 Missing and 1 partial ⚠️

❗ Your organization needs to install the Codecov GitHub app to enable full functionality.

Additional details and impacted files

@@            Coverage Diff             @@
##           master     #843      +/-   ##
==========================================
- Coverage   75.91%   75.78%   -0.13%     
==========================================
  Files         226      226              
  Lines       10938    10962      +24     
==========================================
+ Hits         8304     8308       +4     
- Misses       1902     1917      +15     
- Partials      732      737       +5     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[bobvawter]

Reviewed 1 of 6 files at r3, 13 of 13 files at r6, all commit messages.
Reviewable status: all files reviewed, 6 unresolved discussions (waiting on @BramGruneir and @sravotto)

---

internal/source/kafka/consumer.go line 47 at r6 (raw file):

	schema    ident.Schema      // The target schema.
	timeRange hlc.Range         // The time range for incoming mutations.
	fromState []*partitionState // The initial offsets for each partitions.

Sort order.

---

internal/source/kafka/consumer.go line 69 at r6 (raw file):

		session.MarkOffset(marker.topic, marker.partition, marker.offset, "start")
	}
	c.mu.done = make(map[string]bool)

Not knowing how Setup is called, and from which goroutine(s), you should go ahead and lock the mutex to avoid any surprises later on.

---

internal/source/kafka/consumer.go line 110 at r6 (raw file):

			payload, err := c.accumulate(toProcess, message)
			if err != nil {
				log.Error(err)

Use log.WithError(err).Error("message") Otherwise, the error object isn't going to be presented to the logging hooks as an error object.

---

internal/source/kafka/injector.go line 27 at r6 (raw file):

	"github.com/cockroachdb/replicator/internal/conveyor"
	scriptRuntime "github.com/cockroachdb/replicator/internal/script"
	"github.com/cockroachdb/replicator/internal/sequencer/retire"

Did you run crlfmt? I think you'd get a code-quality warning if the imports aren't sorted.

go run github.com/cockroachdb/crlfmt -w -ignore _gen.go .

---

internal/source/kafka/injector.go line 53 at r6 (raw file):

		tgt.Set,
		retire.Set,
		conveyor.Set,

Alphabetize.

---

internal/source/kafka/provider.go line 30 at r6 (raw file):

	ProvideConn,
	ProvideEagerConfig,
	ProvideConveyorConfig,

Sort.

[sravotto]

Reviewable status: 10 of 13 files reviewed, 6 unresolved discussions (waiting on @bobvawter and @BramGruneir)

---

internal/source/kafka/consumer.go line 47 at r6 (raw file):

Previously, bobvawter (Bob Vawter) wrote…

Sort order.

Done.

---

internal/source/kafka/consumer.go line 69 at r6 (raw file):

Previously, bobvawter (Bob Vawter) wrote…

Not knowing how Setup is called, and from which goroutine(s), you should go ahead and lock the mutex to avoid any surprises later on.

Done.

---

internal/source/kafka/consumer.go line 110 at r6 (raw file):

Previously, bobvawter (Bob Vawter) wrote…

Use log.WithError(err).Error("message") Otherwise, the error object isn't going to be presented to the logging hooks as an error object.

Done.

---

internal/source/kafka/injector.go line 27 at r6 (raw file):

Previously, bobvawter (Bob Vawter) wrote…

Did you run crlfmt? I think you'd get a code-quality warning if the imports aren't sorted.

go run github.com/cockroachdb/crlfmt -w -ignore _gen.go .

it seems to me that they are properly sorted.

---

internal/source/kafka/injector.go line 53 at r6 (raw file):

Previously, bobvawter (Bob Vawter) wrote…

Alphabetize.

Done.

---

internal/source/kafka/provider.go line 30 at r6 (raw file):

Previously, bobvawter (Bob Vawter) wrote…

Sort.

Done.

[bobvawter]

Reviewed 3 of 3 files at r7, all commit messages.
Reviewable status: all files reviewed, 1 unresolved discussion (waiting on @BramGruneir and @sravotto)

---

internal/source/kafka/conn_test.go line 40 at r7 (raw file):

		done       bool
		timestamps ident.Map[hlc.Time]
		ensure     ident.Map[bool]

Sort.

[sravotto]

Thanks!

Reviewable status: 12 of 13 files reviewed, 1 unresolved discussion (waiting on @bobvawter and @BramGruneir)

---

internal/source/kafka/conn_test.go line 40 at r7 (raw file):

Previously, bobvawter (Bob Vawter) wrote…

Sort.

Done.