changefeedccl: memory opt in sarama metadata fetching

[wenyihu6]

Prior to this commit, sarama clients fetch metadata for all topics periodically.
This takes a substantial amount of memory when users create too many topics.
This patch changes sarama configuration so that we only fetch metadata for a
minimal set of topics that have been necessary.

Fixes: #113576
Release Note: none

[blathers-crl]

It looks like your PR touches production code but doesn't add or edit any test code. Did you consider adding tests to your PR?

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

[cockroach-teamcity]

This change is 

[wenyihu6]

pkg/ccl/changefeedccl/sink_kafka.go line 265 at r3 (raw file):

			// RefreshMetadata manually to check for any connection error.
			// TODO(wenyihu6): check if this is what we want
			_ = client.Close()

I added this part of changes as a separate commit since I'm not sure if we want this. We can discuss more during our 1:1 today if it is complex.

[miretskiy]

Blathers is wrong -- nice job on writing the roachtest.

I have mostly nits/suggestions on how to improve the test. Good job on this!

Reviewed 1 of 1 files at r2.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @jayshrivastava and @wenyihu6)

---

pkg/ccl/changefeedccl/sink_kafka.go line 265 at r3 (raw file):

Previously, wenyihu6 (Wenyi Hu) wrote…

I added this part of changes as a separate commit since I'm not sure if we want this. We can discuss more during our 1:1 today if it is complex.

I think we need this -- otherwise, I think that the Dial() would actually be useless
as no network requests will be made and thus we won't know about
connectivity errors.

Let's return the error -- this is consistent with what was happening before.
NewClient code in sarama does:

if conf.Metadata.Full {  
  // do an initial fetch of all cluster metadata by specifying an empty list of topics  
  err := client.RefreshMetadata()  
  if err == nil {  
  } else if errors.Is(err, ErrLeaderNotAvailable) || errors.Is(err, ErrReplicaNotAvailable) || errors.Is(err, ErrTopicAuthorizationFailed) || errors.Is(err, ErrClusterAuthorizationFailed) {  
   // indicates that maybe part of the cluster is down, but is not fatal to creating the client  
   Logger.Println(err)  
  } else {  
   close(client.closed) // we haven't started the background updater yet, so we have to do this manually  
   \_ = client.Close()  
   return nil, err  
  }  
}

That is we call RefreshMetadata and sometimes we would return an error which we would treat as
as error to return to the caller. Thus we should do the same.

You also do not need to ignore client.Close() error here; instead you can do:

if err := client.RefreshMetadata(); err != nil {
   return errors.CombineErrors(err, client.Close())
}

---

pkg/ccl/changefeedccl/sink_kafka.go line 1085 at r3 (raw file):

	config.Producer.Return.Successes = true
	config.Producer.Partitioner = newChangefeedPartitioner
	config.Metadata.Full = false

I know we don't have comments on the other params, but let's start to make things better.
Let's add a short comment, something like // do not fetch metadata for all topics, just the ones we need

---

pkg/cmd/roachtest/tests/cdc.go line 1484 at r3 (raw file):

		Cluster:          r.MakeClusterSpec(4, spec.Arch(vm.ArchAMD64)),
		Leases:           registry.MetamorphicLeases,
		CompatibleClouds: registry.AllExceptAWS,

why except AWS? I think this should work on all clouds?
I think this is a copy-paste issue. It's fine to leave it as is since this is consistent with
other tests, but I have filed #115093

---

pkg/cmd/roachtest/tests/cdc.go line 1504 at r3 (raw file):

			t.Status("creating kafka topics")
			for i := 0; i < 1000; i++ {
				if err := kafka.createTopic(ctx, "topicname"+fmt.Sprintf("%v", i)); err != nil {

use "%d" format; or if you prefer, strconv.Itoa(i)

---

pkg/cmd/roachtest/tests/cdc.go line 1521 at r3 (raw file):

			results, err := ct.cluster.RunWithDetails(ct.ctx, t.L(),
				ct.cluster.Range(1, c.Spec().NodeCount-1),
				"grep \"client/metadata fetching metadata for all topics from broker\" logs/cockroach.log")

I think we should also verify that the log contains the message we do expect to see.
I think you can do this by simplifying your grep pattern: Grep for
client/metadata fetching metadata for. This will match both the thing we expect
not to see -- client/metadata fetching metadata for all topics -- as well as the thing
we expect to see client/metadata fetching metadata for [topicname0 topicname1 ....].

So, after you got the grep output, do the error checking and then make sure that
Stdout does not contain the thing we don't want to see and contains the thing we do want
to see.

---

pkg/cmd/roachtest/tests/cdc.go line 1532 at r3 (raw file):

					t.Fatalf("expected empty stdout but got %s", res.Stdout)
				}
				if res.Stderr != "" {

Probably not needed -- grep prints on stdout; and if anything is printed on stderr, it should also exit with error code (and if it's some sort of an "error info" message -- I don't know, like "disk full" -- we don't care about that).

[miretskiy]

I think you should squash the commits together.

[wenyihu6]

pkg/ccl/changefeedccl/sink_kafka.go line 1085 at r3 (raw file):

Previously, miretskiy (Yevgeniy Miretskiy) wrote…

I know we don't have comments on the other params, but let's start to make things better.
Let's add a short comment, something like // do not fetch metadata for all topics, just the ones we need

Done.

[wenyihu6]

pkg/cmd/roachtest/tests/cdc.go line 1484 at r3 (raw file):

Previously, miretskiy (Yevgeniy Miretskiy) wrote…

why except AWS? I think this should work on all clouds?
I think this is a copy-paste issue. It's fine to leave it as is since this is consistent with
other tests, but I have filed #115093

I will try looking into this issue after I finish my current milestones. From a brief look, it looks like this convention was introduced by 5eb4a53.

[wenyihu6]

pkg/cmd/roachtest/tests/cdc.go line 1504 at r3 (raw file):

Previously, miretskiy (Yevgeniy Miretskiy) wrote…

use "%d" format; or if you prefer, strconv.Itoa(i)

Done.

[wenyihu6]

pkg/cmd/roachtest/tests/cdc.go line 1521 at r3 (raw file):

Previously, wenyihu6 (Wenyi Hu) wrote…

Just realized a chicken and egg situation here - calling Dial() here causes sarama to print client/metadata fetching metadata for all topics from broker again. We could try changing sarama.Logger inside Dial() but it feel unnecessary and might make the code confusing.

Done.

[wenyihu6]

Do you might know why sarama outputs client/metadata fetching metadata for [‹×› ‹×› ‹×› ‹×› ‹×› ‹×› ‹×› ‹×› ‹×›] from broker ‹×› with the ‹×› placeholders rather than the actual strings? I tried printing the string and they were replaced by the actual string client/metadata fetching metadata for [‹customer› ‹history› ‹new_order› ‹district› ‹item› ‹order_line› ‹warehouse› ‹order› ‹stock›] from broker ‹10.142.1.135:9092› and the test passed. But if I do not try printing it, the test fails.

[wenyihu6]

pkg/cmd/roachtest/tests/cdc.go line 1503 at r6 (raw file):

Previously, miretskiy (Yevgeniy Miretskiy) wrote…

does it take long time to create many topics?
nit: consider having const numTopics = ... and use "creating %d kafka topics" status message above.

I cut it to 10 mainly because we are checking to make sure we are not fetching metadata for every topics here. Changed it to 100 now that we only check for the topic name prefix.

[wenyihu6]

pkg/cmd/roachtest/tests/cdc.go line 1530 at r6 (raw file):

Previously, miretskiy (Yevgeniy Miretskiy) wrote…

couldn't you just check to make sure that stdout contains
fmt.Sprintf("fetching metadata for %v topics", allTpccTargets)

(the %v should print the list, space separated, surrounded by [] -- like you see in the logs).

I think the order that the elements of the slice are printed do not seem to be consistent tho.

[wenyihu6]

pkg/cmd/roachtest/tests/cdc.go line 1539 at r6 (raw file):

Previously, miretskiy (Yevgeniy Miretskiy) wrote…

I think checking for all topic names explicitly is probably an overkill.
You can simply take "topicname" and make it into a constant with an "obvious" value -- something like `const ignoreTopicPrefix = "ignore_topic_do_not_fetch". Simply check to see if stdout contains a single instance of that topic prefix.

Agreed. I changed it back to creating 100 topics as well.

[wenyihu6]

pkg/cmd/roachtest/tests/cdc.go line 1532 at r3 (raw file):

Previously, miretskiy (Yevgeniy Miretskiy) wrote…

Probably not needed -- grep prints on stdout; and if anything is printed on stderr, it should also exit with error code (and if it's some sort of an "error info" message -- I don't know, like "disk full" -- we don't care about that).

Ack.

[wenyihu6]

Friendly ping on this review - hoping to merge it this week.

[miretskiy]

Reviewed 1 of 1 files at r7.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @jayshrivastava and @wenyihu6)

---

pkg/ccl/changefeedccl/sink_kafka.go line 260 at r6 (raw file):

Previously, wenyihu6 (Wenyi Hu) wrote…

I think some unit tests failed with accessing nil pointer when I excluded this check.

I suspect that's because we use OverrideClientConfig knob (https://github.com/cockroachdb/cockroach/blob/b7b0b433f8db34eaf59872dfb376187f02f588b6/pkg/ccl/changefeedccl/sink_kafka.go#L277-L277)...

OverrideClientInit: func(config \*sarama.Config) (kafkaClient, error) {  
  return nil, nil  
},

I think it would be better to change those tests to return a non-nil, &fakeKafkaClient{config} instead.

[wenyihu6]

pkg/ccl/changefeedccl/sink_kafka.go line 260 at r6 (raw file):

Previously, miretskiy (Yevgeniy Miretskiy) wrote…

I suspect that's because we use OverrideClientConfig knob (https://github.com/cockroachdb/cockroach/blob/b7b0b433f8db34eaf59872dfb376187f02f588b6/pkg/ccl/changefeedccl/sink_kafka.go#L277-L277)...

OverrideClientInit: func(config \*sarama.Config) (kafkaClient, error) {  
  return nil, nil  
},

I think it would be better to change those tests to return a non-nil, &fakeKafkaClient{config} instead.

Done. Changed it to return fakeKafkaClient and nil error instead.

OverrideClientInit: func(config *sarama.Config) (kafkaClient, error) {
				client := &fakeKafkaClient{config}
				return client, nil
			},

[wenyihu6]

The last push was to fix the comment & rebase on master.

TFTR!

bors r=miretskiy

[craig]

Build failed (retrying...):

• Bazel Essential CI (Cockroach)

[craig]

Build succeeded:

• Bazel Essential CI (Cockroach)