changefeedccl: support alternative kafka hash algo by rharding6373 · Pull Request #161265 · cockroachdb/cockroach

rharding6373 · 2026-01-16T17:25:17Z

Previously, the franz-go based kafka sink only supported the fnv-1a hashing algorithm, because the hashing function was compatible with the original sarama-based sink. However, some users need alternative hashing functions that are compatible with their existing ecosystem.

This PR adds the partition_alg changefeed option, with the options fnv-1a (default) and murmur2 (kafka's default). This is protected by a cluster setting.

Epic: CRDB-58732
Fixes: #161202

Release note (general change): Changefeeds now support the partition_alg
option for specifying a kafka partitioning algorithm. Currently fnv-1a
(default) and murmur2 are supported. The option is only valid on kafka
v2 sinks. This is protected by the cluster setting
changefeed.partition_alg.enabled. An example usage: SET CLUSTER SETTING changefeed.partition_alg.enabled=true; CREATE CHANGEFEED ... INTO 'kafka://...' WITH partition_alg='murmur2'; Note that if a changefeed is
created using the murmur2 algorithm, and then the cluster setting is
disabled, the changefeed will continue using the murmur2 algorithm
unless the changefeed is altered to use a differed partition_alg.

cockroach-teamcity · 2026-01-16T17:25:30Z

This change is

aerfrei

Looks good! Great to have this done. Left some nits

aerfrei · 2026-01-16T17:29:47Z

pkg/ccl/changefeedccl/changefeed_stmt.go

+	if opts.IsSet(changefeedbase.OptHashAlg) {
+		if !changefeedbase.HashAlgEnabled.Get(&settings.SV) {
+			return errors.Newf(
+				"option %s requires cluster setting %s to be enabled",


nit: We might want to add a test for this error message.

aerfrei · 2026-01-16T17:30:14Z

pkg/ccl/changefeedccl/changefeed_stmt.go

 		}
 	}
+
+	// Check feature flag for hash_alg.


nit: This comment seems to mostly read aloud the code below, maybe remove it?

aerfrei · 2026-01-16T17:34:24Z

pkg/ccl/changefeedccl/changefeedbase/settings.go

+	settings.ApplicationLevel,
+	"changefeed.hash_alg.enabled",
+	"if enabled, allows specifying the hash_alg changefeed option to"+
+		" choose between fnv-1a (default) and murmur2 hash functions for"+


nit: The strings we're concatenating here do not have a space between the second and third, meaning this will become "... forKafka partitioning". To match what we do elsewhere in this file, have these end with "+ and no space at the beginning of the strings.

aerfrei · 2026-01-16T17:39:35Z

pkg/ccl/changefeedccl/sink_kafka_v2.go

+	var inner kgo.Partitioner
+
+	if changefeedbase.HashAlgEnabled.Get(&settings.SV) && strings.ToLower(hashMethod) == "murmur2" {
+		// Use franz-go default (murmur2)


super nit: this comment and the one below "Default to fnv-1a" could be full sentences. Maybe we'd prefer something like "murmur2 is the franz-go default and can be specified by the changefeed". And "When not otherwise specified, we use the Sarama default hash method of fnv-1a."

Did some word smithing, ptal.

aerfrei · 2026-01-16T17:42:19Z

pkg/ccl/changefeedccl/sink_test.go

 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)

+	settings := cluster.MakeTestingClusterSettings()


nit: Might we want to define this before we we define kgoPartitioner, where it's used?

aerfrei · 2026-01-16T17:51:15Z

pkg/ccl/changefeedccl/sink_kafka_v2.go

+	if changefeedbase.HashAlgEnabled.Get(&settings.SV) && strings.ToLower(hashMethod) == "murmur2" {
+		// Use franz-go default (murmur2)
+		inner = kgo.StickyKeyPartitioner(nil /* hasher */)
+	} else {


To avoid this else, we should be able to return directly in the if and then handle the default case not nested.

rharding6373

Thanks for the review! Ready for another look.

rharding6373 · 2026-01-16T18:10:03Z

pkg/ccl/changefeedccl/changefeed_stmt.go

 		}
 	}
+
+	// Check feature flag for hash_alg.


rharding6373 · 2026-01-16T18:11:36Z

pkg/ccl/changefeedccl/changefeed_stmt.go

+	if opts.IsSet(changefeedbase.OptHashAlg) {
+		if !changefeedbase.HashAlgEnabled.Get(&settings.SV) {
+			return errors.Newf(
+				"option %s requires cluster setting %s to be enabled",


rharding6373 · 2026-01-16T18:12:21Z

pkg/ccl/changefeedccl/changefeedbase/settings.go

+	settings.ApplicationLevel,
+	"changefeed.hash_alg.enabled",
+	"if enabled, allows specifying the hash_alg changefeed option to"+
+		" choose between fnv-1a (default) and murmur2 hash functions for"+


rharding6373 · 2026-01-16T18:17:15Z

pkg/ccl/changefeedccl/sink_kafka_v2.go

+	var inner kgo.Partitioner
+
+	if changefeedbase.HashAlgEnabled.Get(&settings.SV) && strings.ToLower(hashMethod) == "murmur2" {
+		// Use franz-go default (murmur2)


Did some word smithing, ptal.

rharding6373 · 2026-01-16T18:17:29Z

pkg/ccl/changefeedccl/sink_kafka_v2.go

+	if changefeedbase.HashAlgEnabled.Get(&settings.SV) && strings.ToLower(hashMethod) == "murmur2" {
+		// Use franz-go default (murmur2)
+		inner = kgo.StickyKeyPartitioner(nil /* hasher */)
+	} else {


rharding6373 · 2026-01-16T18:19:06Z

pkg/ccl/changefeedccl/sink_test.go

 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)

+	settings := cluster.MakeTestingClusterSettings()


github-actions · 2026-01-16T18:47:26Z

Potential Bug(s) Detected

The three-stage Claude Code analysis has identified potential bug(s) in this PR that may warrant investigation.

Next Steps:
Please review the detailed findings in the workflow run.

Note: When viewing the workflow output, scroll to the bottom to find the Final Analysis Summary.

After you review the findings, please tag the issue as follows:

If the detected issue is real or was helpful in any way, please tag the issue with O-AI-Review-Real-Issue-Found
If the detected issue was not helpful in any way, please tag the issue with O-AI-Review-Not-Helpful

aerfrei · 2026-01-16T18:51:20Z

pkg/ccl/changefeedccl/sink_kafka_v2.go

+		inner = kgo.StickyKeyPartitioner(nil /* hasher */)
+		return &kgoChangefeedPartitioner{inner: inner}
+	}
+	// If not enabled, or if fnv-1a is specified, use fnv-1a by default.


nit: Maybe "If the cluster setting is not enabled or no option is specified, use fnv-1a by default.". It wasn't clear to me what is not enabled (the cluster setting? or no option provided?). And if we specify fnv-1a, it's not really by default.

aerfrei · 2026-01-16T18:56:31Z

pkg/ccl/changefeedccl/sink_kafka_v2.go

+func newKgoChangefeedPartitioner(hashMethod string, settings *cluster.Settings) kgo.Partitioner {
+	var inner kgo.Partitioner
+
+	if changefeedbase.HashAlgEnabled.Get(&settings.SV) && strings.ToLower(hashMethod) == "murmur2" {


Claude raised an issue here that seems legit to me: we actually don't want to check this cluster setting here, since if the setting was turned on to create the changefeed successfully with murmur2, if it is turned off later we don't want to switch to the fnv-1a hash mid changefeed. We should also probably add a test for this scenario if possible.

Location: pkg/ccl/changefeedccl/sink_kafka_v2.go:620

Issue: The runtime cluster setting check in newKgoChangefeedPartitioner causes partition inconsistency for existing changefeeds. When a changefeed created with hash_alg='murmur2' resumes after the cluster setting changefeed.hash_alg.enabled is disabled, the partitioner silently falls back to fnv-1a, routing the same keys to different Kafka partitions.

Hm, this is an interesting case where using a flag to prevent executing a risky new codepath (I realize the new codepath is not very risky, but this is why we require flags when backporting), can lead to weird, possibly undesirable behavior.

I removed the flag from the partitioner code, since it doesn't seem risky. I updated the commit message to reflect the expected behavior if the setting is toggled.

I tested manually to make sure it behaves as expected. Unfortunately there's a blocker to implementing a test in the test suite for this case, since kafkaFeed does not support multiple partitions, which are critical for validating that the hasher is hashing keys correctly. I spent a couple hours on this today but it's going to take more effort. I created #161310 to track this effort.

Manual testing output: https://gist.github.com/rharding6373/159b952498435f0248e459c224f502bd

Great, I changed the PR description to reflect the commit message change. I'm thinking with that manual testing even without the work of #161310 it should be fine to merge. Approving.

pkg/ccl/changefeedccl/changefeed_stmt.go

pkg/ccl/changefeedccl/changefeedbase/options.go

andyyang890 · 2026-01-20T20:57:44Z

pkg/ccl/changefeedccl/changefeed_stmt.go

 	}
+
+	if opts.IsSet(changefeedbase.OptHashAlg) {
+		if !changefeedbase.HashAlgEnabled.Get(&settings.SV) {


We should do the cluster setting check in ALTER CHANGEFEED too. We might also want to restrict changing the option altogether so that we don't send keys to different partitions after changing the hash algorithm.

Once you add the check, could you also add a test for it?

We might also want to restrict changing the option altogether so that we don't send keys to different partitions after changing the hash algorithm.

Changing the key partition mapping is the point of changing the hash algorithm. Seems like this should be left to the user, though it also seems like a potential foot-gun.

Added a test for alter changefeed, thanks for catching the lack of coverage. The test shows that the existing check already covers the alter case.

pkg/ccl/changefeedccl/changefeed_stmt.go

pkg/ccl/changefeedccl/sink_kafka_v2.go

rharding6373 · 2026-01-20T23:39:02Z

Changed the name from hash_alg to partition_alg. PTAL.

github-actions · 2026-01-20T23:40:49Z

Potential Bug(s) Detected

The three-stage Claude Code analysis has identified potential bug(s) in this PR that may warrant investigation.

Next Steps:
Please review the detailed findings in the workflow run.

Note: When viewing the workflow output, scroll to the bottom to find the Final Analysis Summary.

After you review the findings, please tag the issue as follows:

If the detected issue is real or was helpful in any way, please tag the issue with O-AI-Review-Real-Issue-Found
If the detected issue was not helpful in any way, please tag the issue with O-AI-Review-Not-Helpful

Previously, the franz-go based kafka sink only supported the fnv-1a hashing algorithm, because the hashing function was compatible with the original sarama-based sink. However, some users need alternative hashing functions that are compatible with their existing ecosystem. This PR adds the `partition_alg` changefeed option, with the options `fnv-1a` (default) and `murmur2` (kafka's default). This is protected by a cluster setting. Epic: CRDB-58732 Fixes: cockroachdb#161202 Release note (general change): Changefeeds now support the `partition_alg` option for specifying a kafka partitioning algorithm. Currently `fnv-1a` (default) and `murmur2` are supported. The option is only valid on kafka v2 sinks. This is protected by the cluster setting `changefeed.partition_alg.enabled`. An example usage: `SET CLUSTER SETTING changefeed.partition_alg.enabled=true; CREATE CHANGEFEED ... INTO 'kafka://...' WITH partition_alg='murmur2';` Note that if a changefeed is created using the murmur2 algorithm, and then the cluster setting is disabled, the changefeed will continue using the murmur2 algorithm unless the changefeed is altered to use a differed `partition_alg`.

rharding6373 · 2026-01-20T23:49:52Z

Thanks Claude for finding the two instances of the old name I forgot to replace in the PR description.

andyyang890

LGTM!

rharding6373 · 2026-01-21T16:15:50Z

TFTRs!

bors t=aerfrei,andyyang890

andyyang890 · 2026-01-21T16:40:48Z

bors r=aerfrei,andyyang890

(fixing a typo)

craig · 2026-01-21T17:22:47Z

Build succeeded:

blathers-crl · 2026-01-21T17:22:54Z

Based on the specified backports for this PR, I applied new labels to the following linked issue(s). Please adjust the labels as needed to match the branches actually affected by the issue(s), including adding any known older branches.

Issue #161202: branch-release-25.2, branch-release-25.4, branch-release-26.1.

_{🦉 Hoot! I am a Blathers, a bot for CockroachDB. My owner is dev-inf.}

blathers-crl · 2026-01-21T17:23:04Z

Encountered an error creating backports. Some common things that can go wrong:

The backport branch might have already existed.
There was a merge conflict.
The backport branch contained merge commits.

You might need to create your backport manually using the backport tool.

error creating merge commit from b5f89d9 to blathers/backport-release-25.2-161265: POST https://api.github.com/repos/rharding6373/cockroach/merges: 409 Merge conflict []

you may need to manually resolve merge conflicts with the backport tool.

Backport to branch 25.2.x failed. See errors above.

error creating merge commit from b5f89d9 to blathers/backport-release-25.4-161265: POST https://api.github.com/repos/rharding6373/cockroach/merges: 409 Merge conflict []

you may need to manually resolve merge conflicts with the backport tool.

Backport to branch 25.4.x failed. See errors above.

error creating merge commit from b5f89d9 to blathers/backport-release-26.1-161265: POST https://api.github.com/repos/rharding6373/cockroach/merges: 409 Merge conflict []

you may need to manually resolve merge conflicts with the backport tool.

Backport to branch 26.1.x failed. See errors above.

_{🦉 Hoot! I am a Blathers, a bot for CockroachDB. My owner is dev-inf.}

rharding6373 requested a review from aerfrei January 16, 2026 17:25

rharding6373 requested a review from a team as a code owner January 16, 2026 17:25

aerfrei reviewed Jan 16, 2026

View reviewed changes

rharding6373 force-pushed the 20260114-kafka-hashing-option branch from b587855 to 0e147b1 Compare January 16, 2026 18:39

rharding6373 commented Jan 16, 2026

View reviewed changes

github-actions bot added the o-AI-Review-Potential-Issue-Detected AI reviewer found potential issue. Never assign manually—auto-applied by GH action only. label Jan 16, 2026

aerfrei reviewed Jan 16, 2026

View reviewed changes

rharding6373 force-pushed the 20260114-kafka-hashing-option branch from 0e147b1 to 4508455 Compare January 17, 2026 00:09

rharding6373 added the O-AI-Review-Real-Issue-Found AI reviewer found real issue label Jan 17, 2026

rharding6373 force-pushed the 20260114-kafka-hashing-option branch from 4508455 to 17010c2 Compare January 17, 2026 00:13

rharding6373 requested a review from aerfrei January 17, 2026 00:13

aerfrei approved these changes Jan 20, 2026

View reviewed changes

rharding6373 added backport-25.2.x Flags PRs that need to be backported to 25.2 backport-25.4.x Flags PRs that need to be backported to 25.4 backport-26.1.x Flags PRs that need to be backported to 26.1 labels Jan 20, 2026

rharding6373 requested a review from andyyang890 January 20, 2026 16:23

andyyang890 reviewed Jan 20, 2026

View reviewed changes

rharding6373 force-pushed the 20260114-kafka-hashing-option branch 3 times, most recently from 85d2788 to 2292bbc Compare January 20, 2026 23:38

rharding6373 requested a review from andyyang890 January 20, 2026 23:39

rharding6373 force-pushed the 20260114-kafka-hashing-option branch from 2292bbc to b5f89d9 Compare January 20, 2026 23:48

andyyang890 approved these changes Jan 21, 2026

View reviewed changes

craig bot merged commit 60d00e7 into cockroachdb:master Jan 21, 2026
29 checks passed

celeste-cockroachdb bot added the target-release-26.2.0 label Jan 21, 2026

blathers-crl bot mentioned this pull request Jan 21, 2026

cdc: support murmur2 as a kafka hashing algorithm #161202

Closed

blathers-crl bot added the backport-failed label Jan 21, 2026

celeste-cockroachdb bot added the v26.2.0-prerelease label Mar 11, 2026

Conversation

rharding6373 commented Jan 16, 2026 • edited by andyyang890 Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

cockroach-teamcity commented Jan 16, 2026

Uh oh!

aerfrei left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rharding6373 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

github-actions bot commented Jan 16, 2026

Potential Bug(s) Detected

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

rharding6373 commented Jan 20, 2026

Uh oh!

github-actions bot commented Jan 20, 2026

Potential Bug(s) Detected

Uh oh!

rharding6373 commented Jan 20, 2026

Uh oh!

rharding6373 commented Jan 16, 2026 •

edited by andyyang890

Loading