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

qbx2 · AHeise · commit ea3a6417e290 · 2024-08-22T12:59:34.000+02:00
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilder.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilder.java
@@ -284,7 +284,7 @@ private static class CachingTopicSelector<IN> implements Function<IN, String>, S
         public String apply(IN in) {
             final String topic = cache.getOrDefault(in, topicSelector.apply(in));
             cache.put(in, topic);
-            if (cache.size() == CACHE_RESET_SIZE) {
+            if (cache.size() >= CACHE_RESET_SIZE) {
                 cache.clear();
             }
             return topic;

Original file line number	Diff line number	Diff line change
`@@ -284,7 +284,7 @@ private static class CachingTopicSelector<IN> implements Function<IN, String>, S`
`284`	`284`	`public String apply(IN in) {`
`285`	`285`	`final String topic = cache.getOrDefault(in, topicSelector.apply(in));`
`286`	`286`	`cache.put(in, topic);`
`287`		`- if (cache.size() == CACHE_RESET_SIZE) {`
	`287`	`+ if (cache.size() >= CACHE_RESET_SIZE) {`
`288`	`288`	`cache.clear();`
`289`	`289`	`}`
`290`	`290`	`return topic;`