You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2025/11/19 09:15:15 UTC
(flink-connector-kafka) branch v3.3 updated (06159a21 -> 9f631be2)
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a change to branch v3.3
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
from 06159a21 [FLINK-36595][docs] Explicitly set connector compatibility as string to prevent version comparison mismatch
new 26b2a3c0 [FLINK-37281] Fix KafkaUtil#checkProducerLeak
new 4e2007ba [hotfix] Fix leaks in FlinkKafkaProducerTest
new 9f631be2 [hotfix] Fix Python CI
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../flink/connector/kafka/testutils/KafkaUtil.java | 10 +++++--
.../connectors/kafka/FlinkKafkaProducerTest.java | 34 ++++++++++++++--------
flink-python/setup.py | 12 +++++++-
3 files changed, 41 insertions(+), 15 deletions(-)
(flink-connector-kafka) 02/03: [hotfix] Fix leaks in FlinkKafkaProducerTest
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch v3.3
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
commit 4e2007ba41b5389f974c9a612a1a7b17166fce17
Author: Arvid Heise <ar...@apache.org>
AuthorDate: Sun Feb 16 21:01:32 2025 +0100
[hotfix] Fix leaks in FlinkKafkaProducerTest
(cherry picked from commit 2e652a92e48df67d77e04f7aebcf8e97c511b1cb)
---
.../connectors/kafka/FlinkKafkaProducerTest.java | 34 ++++++++++++++--------
1 file changed, 22 insertions(+), 12 deletions(-)
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerTest.java b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerTest.java
index 6fedcc43..5c2d3803 100644
--- a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerTest.java
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerTest.java
@@ -26,6 +26,7 @@ import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartiti
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.kafka.clients.producer.ProducerRecord;
+import org.junit.After;
import org.junit.Test;
import javax.annotation.Nullable;
@@ -33,28 +34,35 @@ import javax.annotation.Nullable;
import java.util.Optional;
import java.util.Properties;
+import static org.apache.flink.connector.kafka.testutils.KafkaUtil.checkProducerLeak;
import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link FlinkKafkaProducer}. */
public class FlinkKafkaProducerTest {
+ @After
+ public void checkLeaks() {
+ checkProducerLeak();
+ }
+
@Test
public void testOpenSerializationSchemaProducer() throws Exception {
OpenTestingSerializationSchema schema = new OpenTestingSerializationSchema();
FlinkKafkaProducer<Integer> kafkaProducer =
new FlinkKafkaProducer<>("localhost:9092", "test-topic", schema);
- OneInputStreamOperatorTestHarness<Integer, Object> testHarness =
+ try (OneInputStreamOperatorTestHarness<Integer, Object> testHarness =
new OneInputStreamOperatorTestHarness<>(
new StreamSink<>(kafkaProducer),
1,
1,
0,
IntSerializer.INSTANCE,
- new OperatorID(1, 1));
+ new OperatorID(1, 1))) {
- testHarness.open();
+ testHarness.open();
- assertThat(schema.openCalled).isTrue();
+ assertThat(schema.openCalled).isTrue();
+ }
}
@Test
@@ -69,18 +77,19 @@ public class FlinkKafkaProducerTest {
properties,
FlinkKafkaProducer.Semantic.AT_LEAST_ONCE);
- OneInputStreamOperatorTestHarness<Integer, Object> testHarness =
+ try (OneInputStreamOperatorTestHarness<Integer, Object> testHarness =
new OneInputStreamOperatorTestHarness<>(
new StreamSink<>(kafkaProducer),
1,
1,
0,
IntSerializer.INSTANCE,
- new OperatorID(1, 1));
+ new OperatorID(1, 1))) {
- testHarness.open();
+ testHarness.open();
- assertThat(schema.openCalled).isTrue();
+ assertThat(schema.openCalled).isTrue();
+ }
}
@Test
@@ -95,18 +104,19 @@ public class FlinkKafkaProducerTest {
properties,
Optional.of(partitioner));
- OneInputStreamOperatorTestHarness<Integer, Object> testHarness =
+ try (OneInputStreamOperatorTestHarness<Integer, Object> testHarness =
new OneInputStreamOperatorTestHarness<>(
new StreamSink<>(kafkaProducer),
1,
1,
0,
IntSerializer.INSTANCE,
- new OperatorID(1, 1));
+ new OperatorID(1, 1))) {
- testHarness.open();
+ testHarness.open();
- assertThat(partitioner.openCalled).isTrue();
+ assertThat(partitioner.openCalled).isTrue();
+ }
}
@Test(expected = NullPointerException.class)
(flink-connector-kafka) 03/03: [hotfix] Fix Python CI
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch v3.3
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
commit 9f631be2c0d2b01746df4df94e47a07dde169f2c
Author: Martijn Visser <29...@users.noreply.github.com>
AuthorDate: Tue Nov 18 18:57:07 2025 +0100
[hotfix] Fix Python CI
---
flink-python/setup.py | 12 +++++++++++-
1 file changed, 11 insertions(+), 1 deletion(-)
diff --git a/flink-python/setup.py b/flink-python/setup.py
index 8e788d4e..98dd13e3 100644
--- a/flink-python/setup.py
+++ b/flink-python/setup.py
@@ -81,7 +81,16 @@ def prepare_pyflink_dir():
connector_version = pom_root.findall(
"./{http://maven.apache.org/POM/4.0.0}version")[0].text.replace("-SNAPSHOT", ".dev0")
- flink_dependency = "apache-flink>=" + flink_version
+ # Pin to the minor version (e.g., >=1.19.1,<1.20.0) to allow patch updates
+ # but prevent major/minor version upgrades
+ version_parts = flink_version.split('.')
+ if len(version_parts) >= 2:
+ next_minor = str(int(version_parts[1]) + 1)
+ next_version = version_parts[0] + '.' + next_minor + '.0'
+ flink_dependency = "apache-flink>=" + flink_version + ",<" + next_version
+ else:
+ # Fallback to exact match if version format is unexpected
+ flink_dependency = "apache-flink==" + flink_version
os.makedirs(LIB_PATH)
connector_jar = \
@@ -133,6 +142,7 @@ if in_flink_source:
# Runs the python setup
setup(
+ py_modules=[],
name=PACKAGE_NAME,
version=connector_version,
include_package_data=True,
(flink-connector-kafka) 01/03: [FLINK-37281] Fix KafkaUtil#checkProducerLeak
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch v3.3
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
commit 26b2a3c0bc9448697ac2d26a2c555b2937f988f4
Author: Arvid Heise <ar...@apache.org>
AuthorDate: Tue Feb 11 08:30:35 2025 +0100
[FLINK-37281] Fix KafkaUtil#checkProducerLeak
Since Java 20, Thread.stop fails, so we just need to remember old leaks to avoid failing subsequent tests.
(cherry picked from commit 707ec4c194cdea26a34e5f0c8540621c37175c61)
---
.../org/apache/flink/connector/kafka/testutils/KafkaUtil.java | 10 ++++++++--
1 file changed, 8 insertions(+), 2 deletions(-)
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/testutils/KafkaUtil.java b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/testutils/KafkaUtil.java
index 7bf7bb00..78b344b6 100644
--- a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/testutils/KafkaUtil.java
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/testutils/KafkaUtil.java
@@ -38,6 +38,7 @@ import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
+import java.util.concurrent.ConcurrentSkipListSet;
import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.fail;
@@ -193,12 +194,17 @@ public class KafkaUtil {
.collect(Collectors.toSet());
}
+ private static final Set<Long> KNOWN_LEAKS = new ConcurrentSkipListSet<>();
+
public static void checkProducerLeak() {
List<Map.Entry<Thread, StackTraceElement[]>> leaks = null;
for (int tries = 0; tries < 10; tries++) {
leaks =
Thread.getAllStackTraces().entrySet().stream()
.filter(KafkaUtil::findAliveKafkaThread)
+ .filter(
+ threadEntry ->
+ !KNOWN_LEAKS.contains(threadEntry.getKey().getId()))
.collect(Collectors.toList());
if (leaks.isEmpty()) {
return;
@@ -210,10 +216,10 @@ public class KafkaUtil {
}
for (Map.Entry<Thread, StackTraceElement[]> leak : leaks) {
- leak.getKey().stop();
+ KNOWN_LEAKS.add(leak.getKey().getId());
}
fail(
- "Detected producer leaks:\n"
+ "Detected new producer leaks:\n"
+ leaks.stream()
.map(KafkaUtil::format)
.collect(Collectors.joining("\n\n")));