[FLINK-33201][Connectors/Kafka] Fix memory leak in CachingTopicSelector

[qbx2]

What is the purpose of the change

In the CachingTopicSelector, a memory leak may occur when the internal logic fails to check the cache size due to a race condition. (

flink-connector-kafka/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilder.java

Lines 287 to 289 in d89a082

	if (cache.size() == CACHE_RESET_SIZE) {
	cache.clear();
	}

)
This PR fixes the memory leak by modifying the logic to be more resilient to failure.

Brief change log

Fix memory leak in CachingTopicSelector that can be triggered by race condition.

Verifying this change

• By analyzing a Java heap dump, I identified a memory leak in the CachingTopicSelector. As in the screenshot, cache has 47,769 elements. If the internal logic were functioning correctly, the number of elements should be less than or equal to CACHE_RESET_SIZE (which is 5).
• Since writing unit tests for this type of bug is challenging, I instead applied a hotfix to our production workload. Before applying the hotfix, the memory leak is observed in the workload in 7 days. After applying the patch, the issue is no longer observed.

[boring-cyborg]

Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html)

[MartijnVisser]

@qbx2 Can you please rebase your PR?

[qbx2]

@MartijnVisser Sure, I just rebased it.

[qbx2]

@MartijnVisser Could you please review?

[cnmckee13]

Hey All- We're experiencing some memory issues and wondering if this could be the cause. Will this PR be reviewed soon? @MartijnVisser @tzulitai

[MartijnVisser]

@AHeise Can you take a look?

[AHeise]

LGTM. I will run the test to make sure nothing happens.

[AHeise]

Oh I see the tests can't be run because it's based on an old branch. I'm rebasing myself.