Add kafka client pool

[Cali0707]

Fixes #3397 , #3139

Proposed Changes

• Create a client pool to use to get clients and cluster admins for the ekb control plane
• Use an LRU eviction method to ensure that there are never more than a max number of connections being made

Release Note

Kafka control plane connections are now pooled, rather than recreated.

Docs

[knative-prow]

Skipping CI for Draft Pull Request.
If you want CI signal for your change, please convert it to an actual PR.
You can still manually trigger a test run with /test all

[codecov]

Codecov Report

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

Project coverage is 48.41%. Comparing base (73c369e) to head (d317978).
Report is 37 commits behind head on main.

Files	Patch %	Lines
control-plane/pkg/kafka/clientpool/client.go	8.69%	124 Missing and 2 partials ⚠️
control-plane/pkg/kafka/clientpool/clientpool.go	50.00%	39 Missing and 6 partials ⚠️
...ontrol-plane/pkg/kafka/sarama/batch_get_offsets.go	0.00%	14 Missing ⚠️
control-plane/pkg/kafka/testing/client_mock.go	0.00%	11 Missing ⚠️
control-plane/pkg/kafka/testing/admin_mock.go	62.50%	2 Missing and 1 partial ⚠️
...ane/pkg/reconciler/broker/namespaced_controller.go	0.00%	3 Missing ⚠️
...lane/pkg/reconciler/consumergroup/consumergroup.go	57.14%	2 Missing and 1 partial ⚠️
control-plane/pkg/prober/async_prober.go	60.00%	2 Missing ⚠️
control-plane/pkg/reconciler/broker/broker.go	66.66%	2 Missing ⚠️
control-plane/cmd/kafka-controller/main.go	0.00%	1 Missing ⚠️
... and 2 more

Additional details and impacted files

@@              Coverage Diff              @@
##               main    #3493       +/-   ##
=============================================
- Coverage     62.18%   48.41%   -13.77%     
=============================================
  Files           188      243       +55     
  Lines         12690    18017     +5327     
  Branches        273        0      -273     
=============================================
+ Hits           7891     8723      +832     
- Misses         4187     8560     +4373     
- Partials        612      734      +122     

Flag	Coverage Δ
java-unittests	?

Flags with carried forward coverage won't be shown. Click here to find out more.

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

[Cali0707]

/hold
While I debug the tests

[Cali0707]

/unhold
/cc @pierDipi @creydr @matzew

[Cali0707]

/retest-required

[Cali0707]

/retest-required

[Cali0707]

/cc @pierDipi

[pierDipi]

When does this case happen?

[Cali0707]

Oh I've fixed this now, we can probably remove it. Basically, if you do something like:

client, err := r.GetKafkaClient(...)
defer client.Close()
if err != nil {
   ...
}

Here, if err is not nil, the deferred client.Close() method panics.

I fixed all the instances where this was happening, so I don't think we actually need this now

[pierDipi]

This is unused, why do we need this method?

[Cali0707]

Good catch, this was from an earlier version of the PR. Let me remove it

[pierDipi]

/lgtm
/approve

[knative-prow]

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: Cali0707, pierDipi

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Needs approval from an approver in each of these files:

• OWNERS [Cali0707,pierDipi]

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[pierDipi]

/retest-required