You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Efrat19 (via GitHub)" <gi...@apache.org> on 2026/03/11 10:45:44 UTC
[PR] [FLINK-39234][tests] Wait for topic intialization in KafkaWriterTestBase based tests [flink-connector-kafka]
Efrat19 opened a new pull request, #237:
URL: https://github.com/apache/flink-connector-kafka/pull/237
KafkaWriterFaultToleranceITCase#testFlushExceptionWhenKafkaUnavailable is flaky in CI due to org.apache.kafka.common.errors.NotLeaderOrFollowerException thrown instead of other expected exceptions (NetworkException / TimeoutException). This is because the topic isn't fully initialized, (see https://github.com/apache/flink-connector-kafka/pull/228) and the failure due to the missing topic races actual failures in the closing KAFKA_CONTAINER
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39234][tests] Wait for topic intialization in KafkaWriterTestBase based tests [flink-connector-kafka]
Posted by "Savonitar (via GitHub)" <gi...@apache.org>.
Savonitar commented on code in PR #237:
URL: https://github.com/apache/flink-connector-kafka/pull/237#discussion_r2976717438
##########
flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java:
##########
@@ -139,43 +137,8 @@ private void tryDelete(AdminClient adminClient, String topic) throws Exception {
@Override
public void createTestTopic(
String topic, int numberOfPartitions, int replicationFactor, Properties properties) {
- createNewTopic(topic, numberOfPartitions, replicationFactor, getStandardProperties());
- }
-
- public static void createNewTopic(
- String topic, int numberOfPartitions, int replicationFactor, Properties properties) {
- LOG.info("Creating topic {}", topic);
- try (AdminClient adminClient = AdminClient.create(properties)) {
- NewTopic topicObj = new NewTopic(topic, numberOfPartitions, (short) replicationFactor);
- adminClient.createTopics(Collections.singleton(topicObj)).all().get();
- CommonTestUtils.waitUtil(
- () -> {
- try {
- // Ensure all partitions have a leader elected and logs initialized
- Map<TopicPartition, OffsetSpec> offsetSpecs = new HashMap<>();
- for (int i = 0; i < numberOfPartitions; i++) {
- offsetSpecs.put(
- new TopicPartition(topic, i), OffsetSpec.earliest());
- }
- adminClient
- .listOffsets(offsetSpecs)
- .all()
- .get(REQUEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
- return true;
- } catch (Exception e) {
- LOG.warn(
- "Partitions for topic {} not yet ready to serve requests",
- topic,
- e);
- return false;
- }
- },
- Duration.ofSeconds(30),
- String.format("New topic \"%s\" is not ready within timeout", topicObj));
- } catch (Exception e) {
- e.printStackTrace();
- fail("Create test topic : " + topic + " failed, " + e.getMessage());
- }
+ createNewTopicAndWaitForPartitionAssignment(
+ topic, numberOfPartitions, replicationFactor, properties);
Review Comment:
It sounds good. However, the CI is now failing with `TopicExistsException`, because the previous behavior didn't check the result of
```
admin.createTopics(Collections.singleton(new NewTopic(topic, 10, (short) 1)));
```
while the topic already existed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39234][tests] Wait for topic intialization in KafkaWriterTestBase based tests [flink-connector-kafka]
Posted by "Savonitar (via GitHub)" <gi...@apache.org>.
Savonitar commented on code in PR #237:
URL: https://github.com/apache/flink-connector-kafka/pull/237#discussion_r2976717438
##########
flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java:
##########
@@ -139,43 +137,8 @@ private void tryDelete(AdminClient adminClient, String topic) throws Exception {
@Override
public void createTestTopic(
String topic, int numberOfPartitions, int replicationFactor, Properties properties) {
- createNewTopic(topic, numberOfPartitions, replicationFactor, getStandardProperties());
- }
-
- public static void createNewTopic(
- String topic, int numberOfPartitions, int replicationFactor, Properties properties) {
- LOG.info("Creating topic {}", topic);
- try (AdminClient adminClient = AdminClient.create(properties)) {
- NewTopic topicObj = new NewTopic(topic, numberOfPartitions, (short) replicationFactor);
- adminClient.createTopics(Collections.singleton(topicObj)).all().get();
- CommonTestUtils.waitUtil(
- () -> {
- try {
- // Ensure all partitions have a leader elected and logs initialized
- Map<TopicPartition, OffsetSpec> offsetSpecs = new HashMap<>();
- for (int i = 0; i < numberOfPartitions; i++) {
- offsetSpecs.put(
- new TopicPartition(topic, i), OffsetSpec.earliest());
- }
- adminClient
- .listOffsets(offsetSpecs)
- .all()
- .get(REQUEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
- return true;
- } catch (Exception e) {
- LOG.warn(
- "Partitions for topic {} not yet ready to serve requests",
- topic,
- e);
- return false;
- }
- },
- Duration.ofSeconds(30),
- String.format("New topic \"%s\" is not ready within timeout", topicObj));
- } catch (Exception e) {
- e.printStackTrace();
- fail("Create test topic : " + topic + " failed, " + e.getMessage());
- }
+ createNewTopicAndWaitForPartitionAssignment(
+ topic, numberOfPartitions, replicationFactor, properties);
Review Comment:
It sounds good. However, the CI is red.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39234][tests] Wait for topic intialization in KafkaWriterTestBase based tests [flink-connector-kafka]
Posted by "Efrat19 (via GitHub)" <gi...@apache.org>.
Efrat19 commented on code in PR #237:
URL: https://github.com/apache/flink-connector-kafka/pull/237#discussion_r2971381124
##########
flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java:
##########
@@ -139,43 +137,8 @@ private void tryDelete(AdminClient adminClient, String topic) throws Exception {
@Override
public void createTestTopic(
String topic, int numberOfPartitions, int replicationFactor, Properties properties) {
- createNewTopic(topic, numberOfPartitions, replicationFactor, getStandardProperties());
- }
-
- public static void createNewTopic(
- String topic, int numberOfPartitions, int replicationFactor, Properties properties) {
- LOG.info("Creating topic {}", topic);
- try (AdminClient adminClient = AdminClient.create(properties)) {
- NewTopic topicObj = new NewTopic(topic, numberOfPartitions, (short) replicationFactor);
- adminClient.createTopics(Collections.singleton(topicObj)).all().get();
- CommonTestUtils.waitUtil(
- () -> {
- try {
- // Ensure all partitions have a leader elected and logs initialized
- Map<TopicPartition, OffsetSpec> offsetSpecs = new HashMap<>();
- for (int i = 0; i < numberOfPartitions; i++) {
- offsetSpecs.put(
- new TopicPartition(topic, i), OffsetSpec.earliest());
- }
- adminClient
- .listOffsets(offsetSpecs)
- .all()
- .get(REQUEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
- return true;
- } catch (Exception e) {
- LOG.warn(
- "Partitions for topic {} not yet ready to serve requests",
- topic,
- e);
- return false;
- }
- },
- Duration.ofSeconds(30),
- String.format("New topic \"%s\" is not ready within timeout", topicObj));
- } catch (Exception e) {
- e.printStackTrace();
- fail("Create test topic : " + topic + " failed, " + e.getMessage());
- }
+ createNewTopicAndWaitForPartitionAssignment(
+ topic, numberOfPartitions, replicationFactor, properties);
Review Comment:
Good catch, thanks!
Looking closely I think the change is for the good though
The original `createTestTopic` implementation takes in a `properties` argument but uses `getStandardProperties()` instead, which is a little counterintuitive
```
@Override
public void createTestTopic(
String topic, int numberOfPartitions, int replicationFactor, Properties properties) {
createNewTopic(topic, numberOfPartitions, replicationFactor, getStandardProperties());
```
I assume it was meant to satisfy the contract.
To fix this I think we can change the abstract call in `KafkaTestEnvironment` to default to standardProperties instead of the empty ones.
This will cover all usages except
`DynamicKafkaSourceExternalContext.setupSplits` which sends `properties` - imo the correct behavior would be to use them, I added a note about it in the commit but if you have concerns I can create a separate PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39234][tests] Wait for topic intialization in KafkaWriterTestBase based tests [flink-connector-kafka]
Posted by "Savonitar (via GitHub)" <gi...@apache.org>.
Savonitar commented on PR #237:
URL: https://github.com/apache/flink-connector-kafka/pull/237#issuecomment-4141966944
As we discussed, I've raised and [merged](https://github.com/apache/flink-connector-kafka/pull/242) the prerequisite that should solve the previous problem.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39234][tests] Wait for topic intialization in KafkaWriterTestBase based tests [flink-connector-kafka]
Posted by "Efrat19 (via GitHub)" <gi...@apache.org>.
Efrat19 commented on PR #237:
URL: https://github.com/apache/flink-connector-kafka/pull/237#issuecomment-4062454569
> Btw, by the mentioned flaky CI, did you mean the failed CI in #234?
Yes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39234][tests] Wait for topic intialization in KafkaWriterTestBase based tests [flink-connector-kafka]
Posted by "Efrat19 (via GitHub)" <gi...@apache.org>.
Efrat19 commented on code in PR #237:
URL: https://github.com/apache/flink-connector-kafka/pull/237#discussion_r3069401085
##########
flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java:
##########
@@ -139,43 +137,8 @@ private void tryDelete(AdminClient adminClient, String topic) throws Exception {
@Override
public void createTestTopic(
String topic, int numberOfPartitions, int replicationFactor, Properties properties) {
- createNewTopic(topic, numberOfPartitions, replicationFactor, getStandardProperties());
- }
-
- public static void createNewTopic(
- String topic, int numberOfPartitions, int replicationFactor, Properties properties) {
- LOG.info("Creating topic {}", topic);
- try (AdminClient adminClient = AdminClient.create(properties)) {
- NewTopic topicObj = new NewTopic(topic, numberOfPartitions, (short) replicationFactor);
- adminClient.createTopics(Collections.singleton(topicObj)).all().get();
- CommonTestUtils.waitUtil(
- () -> {
- try {
- // Ensure all partitions have a leader elected and logs initialized
- Map<TopicPartition, OffsetSpec> offsetSpecs = new HashMap<>();
- for (int i = 0; i < numberOfPartitions; i++) {
- offsetSpecs.put(
- new TopicPartition(topic, i), OffsetSpec.earliest());
- }
- adminClient
- .listOffsets(offsetSpecs)
- .all()
- .get(REQUEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
- return true;
- } catch (Exception e) {
- LOG.warn(
- "Partitions for topic {} not yet ready to serve requests",
- topic,
- e);
- return false;
- }
- },
- Duration.ofSeconds(30),
- String.format("New topic \"%s\" is not ready within timeout", topicObj));
- } catch (Exception e) {
- e.printStackTrace();
- fail("Create test topic : " + topic + " failed, " + e.getMessage());
- }
+ createNewTopicAndWaitForPartitionAssignment(
+ topic, numberOfPartitions, replicationFactor, properties);
Review Comment:
https://github.com/apache/flink-connector-kafka/pull/237#pullrequestreview-4094947261
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39234][tests] Wait for topic intialization in KafkaWriterTestBase based tests [flink-connector-kafka]
Posted by "Efrat19 (via GitHub)" <gi...@apache.org>.
Efrat19 commented on code in PR #237:
URL: https://github.com/apache/flink-connector-kafka/pull/237#discussion_r2971381124
##########
flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java:
##########
@@ -139,43 +137,8 @@ private void tryDelete(AdminClient adminClient, String topic) throws Exception {
@Override
public void createTestTopic(
String topic, int numberOfPartitions, int replicationFactor, Properties properties) {
- createNewTopic(topic, numberOfPartitions, replicationFactor, getStandardProperties());
- }
-
- public static void createNewTopic(
- String topic, int numberOfPartitions, int replicationFactor, Properties properties) {
- LOG.info("Creating topic {}", topic);
- try (AdminClient adminClient = AdminClient.create(properties)) {
- NewTopic topicObj = new NewTopic(topic, numberOfPartitions, (short) replicationFactor);
- adminClient.createTopics(Collections.singleton(topicObj)).all().get();
- CommonTestUtils.waitUtil(
- () -> {
- try {
- // Ensure all partitions have a leader elected and logs initialized
- Map<TopicPartition, OffsetSpec> offsetSpecs = new HashMap<>();
- for (int i = 0; i < numberOfPartitions; i++) {
- offsetSpecs.put(
- new TopicPartition(topic, i), OffsetSpec.earliest());
- }
- adminClient
- .listOffsets(offsetSpecs)
- .all()
- .get(REQUEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
- return true;
- } catch (Exception e) {
- LOG.warn(
- "Partitions for topic {} not yet ready to serve requests",
- topic,
- e);
- return false;
- }
- },
- Duration.ofSeconds(30),
- String.format("New topic \"%s\" is not ready within timeout", topicObj));
- } catch (Exception e) {
- e.printStackTrace();
- fail("Create test topic : " + topic + " failed, " + e.getMessage());
- }
+ createNewTopicAndWaitForPartitionAssignment(
+ topic, numberOfPartitions, replicationFactor, properties);
Review Comment:
Good catch, thanks!
Looking closely I think the change is for the good though
This `createTestTopic` implementation takes in a `properties` argument but uses `getStandardProperties()` instead, which is a little counterintuitive
```
@Override
public void createTestTopic(
String topic, int numberOfPartitions, int replicationFactor, Properties properties) {
createNewTopicAndWaitForPartitionAssignment(
topic, numberOfPartitions, replicationFactor, getStandardProperties());
```
I assume it was meant to satisfy the contract.
To fix this I think we can change the abstract call in `KafkaTestEnvironment` to default to standardProperties instead of the empty ones.
This will cover all usages except
`DynamicKafkaSourceExternalContext.setupSplits` which originally sent properties - imo the correct behavior would be to use them, I added a note about it in the commit but if you have concerns I can create a separate PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39234][tests] Wait for topic intialization in KafkaWriterTestBase based tests [flink-connector-kafka]
Posted by "Savonitar (via GitHub)" <gi...@apache.org>.
Savonitar commented on code in PR #237:
URL: https://github.com/apache/flink-connector-kafka/pull/237#discussion_r2961110405
##########
flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java:
##########
@@ -139,43 +137,8 @@ private void tryDelete(AdminClient adminClient, String topic) throws Exception {
@Override
public void createTestTopic(
String topic, int numberOfPartitions, int replicationFactor, Properties properties) {
- createNewTopic(topic, numberOfPartitions, replicationFactor, getStandardProperties());
- }
-
- public static void createNewTopic(
- String topic, int numberOfPartitions, int replicationFactor, Properties properties) {
- LOG.info("Creating topic {}", topic);
- try (AdminClient adminClient = AdminClient.create(properties)) {
- NewTopic topicObj = new NewTopic(topic, numberOfPartitions, (short) replicationFactor);
- adminClient.createTopics(Collections.singleton(topicObj)).all().get();
- CommonTestUtils.waitUtil(
- () -> {
- try {
- // Ensure all partitions have a leader elected and logs initialized
- Map<TopicPartition, OffsetSpec> offsetSpecs = new HashMap<>();
- for (int i = 0; i < numberOfPartitions; i++) {
- offsetSpecs.put(
- new TopicPartition(topic, i), OffsetSpec.earliest());
- }
- adminClient
- .listOffsets(offsetSpecs)
- .all()
- .get(REQUEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
- return true;
- } catch (Exception e) {
- LOG.warn(
- "Partitions for topic {} not yet ready to serve requests",
- topic,
- e);
- return false;
- }
- },
- Duration.ofSeconds(30),
- String.format("New topic \"%s\" is not ready within timeout", topicObj));
- } catch (Exception e) {
- e.printStackTrace();
- fail("Create test topic : " + topic + " failed, " + e.getMessage());
- }
+ createNewTopicAndWaitForPartitionAssignment(
+ topic, numberOfPartitions, replicationFactor, properties);
Review Comment:
Previously, in the old code, it was `getStandardProperties()`.
I see that this method is called from [org.apache.flink.streaming.connectors.kafka.KafkaTestEnvironment](https://github.com/apache/flink-connector-kafka/blob/main/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironment.java#L85), and the call site passes empty properties (`new Properties()`). Then these empty properties are used by `AdminClient`.
Meanwhile, `getStandardProperties()` used properties with `bootstrap.server` and others.
I recommend to keep existing behavior (passing `getStandardProperties()`), unless Im missing something, indeed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39234][tests] Wait for topic intialization in KafkaWriterTestBase based tests [flink-connector-kafka]
Posted by "Savonitar (via GitHub)" <gi...@apache.org>.
Savonitar commented on code in PR #237:
URL: https://github.com/apache/flink-connector-kafka/pull/237#discussion_r2955680643
##########
flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java:
##########
@@ -144,38 +142,7 @@ public void createTestTopic(
public static void createNewTopic(
String topic, int numberOfPartitions, int replicationFactor, Properties properties) {
- LOG.info("Creating topic {}", topic);
- try (AdminClient adminClient = AdminClient.create(properties)) {
- NewTopic topicObj = new NewTopic(topic, numberOfPartitions, (short) replicationFactor);
- adminClient.createTopics(Collections.singleton(topicObj)).all().get();
- CommonTestUtils.waitUtil(
- () -> {
- try {
- // Ensure all partitions have a leader elected and logs initialized
- Map<TopicPartition, OffsetSpec> offsetSpecs = new HashMap<>();
- for (int i = 0; i < numberOfPartitions; i++) {
- offsetSpecs.put(
- new TopicPartition(topic, i), OffsetSpec.earliest());
- }
- adminClient
- .listOffsets(offsetSpecs)
- .all()
- .get(REQUEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
- return true;
- } catch (Exception e) {
- LOG.warn(
- "Partitions for topic {} not yet ready to serve requests",
- topic,
- e);
- return false;
- }
- },
- Duration.ofSeconds(30),
- String.format("New topic \"%s\" is not ready within timeout", topicObj));
- } catch (Exception e) {
- e.printStackTrace();
- fail("Create test topic : " + topic + " failed, " + e.getMessage());
- }
+ createNewTopicAndWaitForPartitionAssignment(topic, numberOfPartitions, replicationFactor, properties);
Review Comment:
What is the motivation to keep it not inlined?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39234][tests] Wait for topic intialization in KafkaWriterTestBase based tests [flink-connector-kafka]
Posted by "Efrat19 (via GitHub)" <gi...@apache.org>.
Efrat19 commented on code in PR #237:
URL: https://github.com/apache/flink-connector-kafka/pull/237#discussion_r2971381124
##########
flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java:
##########
@@ -139,43 +137,8 @@ private void tryDelete(AdminClient adminClient, String topic) throws Exception {
@Override
public void createTestTopic(
String topic, int numberOfPartitions, int replicationFactor, Properties properties) {
- createNewTopic(topic, numberOfPartitions, replicationFactor, getStandardProperties());
- }
-
- public static void createNewTopic(
- String topic, int numberOfPartitions, int replicationFactor, Properties properties) {
- LOG.info("Creating topic {}", topic);
- try (AdminClient adminClient = AdminClient.create(properties)) {
- NewTopic topicObj = new NewTopic(topic, numberOfPartitions, (short) replicationFactor);
- adminClient.createTopics(Collections.singleton(topicObj)).all().get();
- CommonTestUtils.waitUtil(
- () -> {
- try {
- // Ensure all partitions have a leader elected and logs initialized
- Map<TopicPartition, OffsetSpec> offsetSpecs = new HashMap<>();
- for (int i = 0; i < numberOfPartitions; i++) {
- offsetSpecs.put(
- new TopicPartition(topic, i), OffsetSpec.earliest());
- }
- adminClient
- .listOffsets(offsetSpecs)
- .all()
- .get(REQUEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
- return true;
- } catch (Exception e) {
- LOG.warn(
- "Partitions for topic {} not yet ready to serve requests",
- topic,
- e);
- return false;
- }
- },
- Duration.ofSeconds(30),
- String.format("New topic \"%s\" is not ready within timeout", topicObj));
- } catch (Exception e) {
- e.printStackTrace();
- fail("Create test topic : " + topic + " failed, " + e.getMessage());
- }
+ createNewTopicAndWaitForPartitionAssignment(
+ topic, numberOfPartitions, replicationFactor, properties);
Review Comment:
Good catch, thanks!
Looking closely I think the change is for the good though
The original `createTestTopic` implementation takes in a `properties` argument but uses `getStandardProperties()` instead, which is a little counterintuitive
```
@Override
public void createTestTopic(
String topic, int numberOfPartitions, int replicationFactor, Properties properties) {
createNewTopic(topic, numberOfPartitions, replicationFactor, getStandardProperties());
```
I assume it was meant to satisfy the contract.
To fix this I think we can change the abstract call in `KafkaTestEnvironment` to default to standardProperties instead of the empty ones.
This will cover all usages except
`DynamicKafkaSourceExternalContext.setupSplits` which originally sent properties - imo the correct behavior would be to use them, I added a note about it in the commit but if you have concerns I can create a separate PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39234][tests] Wait for topic intialization in KafkaWriterTestBase based tests [flink-connector-kafka]
Posted by "Efrat19 (via GitHub)" <gi...@apache.org>.
Efrat19 commented on PR #237:
URL: https://github.com/apache/flink-connector-kafka/pull/237#issuecomment-4088111517
> Thanks for the fix. Looks like there are formatting related errors.
Ah probably due to https://github.com/apache/flink-connector-kafka/pull/233
Let me rebase
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39234][tests] Wait for topic intialization in KafkaWriterTestBase based tests [flink-connector-kafka]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui merged PR #237:
URL: https://github.com/apache/flink-connector-kafka/pull/237
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org