You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2021/10/01 13:09:13 UTC
[GitHub] [flink] fapaul opened a new pull request #17406: [FLINK-24405][tests] Introduce util to reliably drain all messages from a kafka topic
fapaul opened a new pull request #17406:
URL: https://github.com/apache/flink/pull/17406
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
The approach used before the introduction was to poll the topic with a
given timeout and if the poll did not return a record treat it as all
records have been read. Unfortunately, due to network issues or similar
the poll can easily return without any records but end offset was not
reached making the tests unreliable. The new util always drains all
messages until the end offset is reached.
## Brief change log
- Migrate all kafka tests reading the topic content to the new util
## Verifying this change
- Existing tests still pass
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**)
- The serializers: (yes / **no** / don't know)
- The runtime per-record code paths (performance sensitive): (yes / **no** / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / **no** / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (yes / **no**)
- If yes, how is the feature documented? (**not applicable** / docs / JavaDocs / not documented)
--
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
[GitHub] [flink] flinkbot commented on pull request #17406: [FLINK-24405][tests] Introduce util to reliably drain all messages from a kafka topic
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #17406:
URL: https://github.com/apache/flink/pull/17406#issuecomment-932253575
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "21ea516945a020245a9699d6bed90e0d3c4cb2b7",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "21ea516945a020245a9699d6bed90e0d3c4cb2b7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 21ea516945a020245a9699d6bed90e0d3c4cb2b7 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] fapaul commented on a change in pull request #17406: [FLINK-24405][tests] Introduce util to reliably drain all messages from a kafka topic
Posted by GitBox <gi...@apache.org>.
fapaul commented on a change in pull request #17406:
URL: https://github.com/apache/flink/pull/17406#discussion_r722243031
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
##########
@@ -279,32 +279,16 @@ public String getVersion() {
}
@Override
+ @SuppressWarnings("unchecked")
public <K, V> Collection<ConsumerRecord<K, V>> getAllRecordsFromTopic(
- Properties properties, String topic, int partition, long timeout) {
- List<ConsumerRecord<K, V>> result = new ArrayList<>();
-
- try (KafkaConsumer<K, V> consumer = new KafkaConsumer<>(properties)) {
- consumer.assign(Arrays.asList(new TopicPartition(topic, partition)));
-
- while (true) {
- boolean processedAtLeastOneRecord = false;
-
- // wait for new records with timeout and break the loop if we didn't get any
- Iterator<ConsumerRecord<K, V>> iterator = consumer.poll(timeout).iterator();
- while (iterator.hasNext()) {
- ConsumerRecord<K, V> record = iterator.next();
- result.add(record);
- processedAtLeastOneRecord = true;
- }
-
- if (!processedAtLeastOneRecord) {
- break;
- }
- }
- consumer.commitSync();
- }
-
- return UnmodifiableList.decorate(result);
+ Properties properties, String topic) {
+ final boolean committed =
+ Objects.equals(
+ properties.getProperty(
+ ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_uncommitted"),
+ "read_committed");
+ return UnmodifiableList.decorate(
+ KafkaUtil.drainAllRecordsFromTopic(topic, properties, committed));
Review comment:
Hmm I would like to leave it as it is to not leave all configuration options because I currently cannot really reason about the different things people want to set.
--
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
[GitHub] [flink] fapaul commented on a change in pull request #17406: [FLINK-24405][tests] Introduce util to reliably drain all messages from a kafka topic
Posted by GitBox <gi...@apache.org>.
fapaul commented on a change in pull request #17406:
URL: https://github.com/apache/flink/pull/17406#discussion_r722179479
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaUtil.java
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.kafka.sink;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/** Collection of methods to interact with a Kafka cluster. */
+public class KafkaUtil {
+
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaUtil.class);
+ private static final Duration CONSUMER_POLL_DURATION = Duration.ofSeconds(1);
+
+ private KafkaUtil() {}
+
+ /**
+ * Drain all records available from the given topic from the beginning until the current highest
+ * offset.
+ *
+ * <p>This method will fetch the latest offsets for the partitions once and only return records
+ * until that point.
+ *
+ * @param topic to fetch from
+ * @param properties used to configure the created {@link KafkaConsumer}
+ * @param committed determines the mode {@link ConsumerConfig#ISOLATION_LEVEL_CONFIG} with which
+ * the consumer reads the records.
+ * @return all {@link ConsumerRecord} in the topic
+ * @throws KafkaException
+ */
+ public static List<ConsumerRecord<byte[], byte[]>> drainAllRecordsFromTopic(
+ String topic, Properties properties, boolean committed) throws KafkaException {
Review comment:
some of the old tests use different serializers and whether the offsets are committed automatically.
--
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
[GitHub] [flink] AHeise merged pull request #17406: [FLINK-24405][tests] Introduce util to reliably drain all messages from a kafka topic
Posted by GitBox <gi...@apache.org>.
AHeise merged pull request #17406:
URL: https://github.com/apache/flink/pull/17406
--
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
[GitHub] [flink] flinkbot commented on pull request #17406: [FLINK-24405][tests] Introduce util to reliably drain all messages from a kafka topic
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #17406:
URL: https://github.com/apache/flink/pull/17406#issuecomment-932224729
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit 7a2145743176bf71fbc1d7abaac55cdd9019c258 (Fri Oct 01 13:23:11 UTC 2021)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
--
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
[GitHub] [flink] AHeise commented on a change in pull request #17406: [FLINK-24405][tests] Introduce util to reliably drain all messages from a kafka topic
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #17406:
URL: https://github.com/apache/flink/pull/17406#discussion_r721318188
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaUtil.java
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.kafka.sink;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/** Collection of methods to interact with a Kafka cluster. */
+public class KafkaUtil {
+
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaUtil.class);
+ private static final Duration CONSUMER_POLL_DURATION = Duration.ofSeconds(1);
+
+ private KafkaUtil() {}
+
+ /**
+ * Drain all records available from the given topic from the beginning until the current highest
+ * offset.
+ *
+ * <p>This method will fetch the latest offsets for the partitions once and only return records
+ * until that point.
+ *
+ * @param topic to fetch from
+ * @param properties used to configure the created {@link KafkaConsumer}
+ * @param committed determines the mode {@link ConsumerConfig#ISOLATION_LEVEL_CONFIG} with which
+ * the consumer reads the records.
+ * @return all {@link ConsumerRecord} in the topic
+ * @throws KafkaException
+ */
+ public static List<ConsumerRecord<byte[], byte[]>> drainAllRecordsFromTopic(
+ String topic, Properties properties, boolean committed) throws KafkaException {
Review comment:
What's in `properties` except the URL at this point?
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
##########
@@ -279,32 +279,16 @@ public String getVersion() {
}
@Override
+ @SuppressWarnings("unchecked")
public <K, V> Collection<ConsumerRecord<K, V>> getAllRecordsFromTopic(
- Properties properties, String topic, int partition, long timeout) {
- List<ConsumerRecord<K, V>> result = new ArrayList<>();
-
- try (KafkaConsumer<K, V> consumer = new KafkaConsumer<>(properties)) {
- consumer.assign(Arrays.asList(new TopicPartition(topic, partition)));
-
- while (true) {
- boolean processedAtLeastOneRecord = false;
-
- // wait for new records with timeout and break the loop if we didn't get any
- Iterator<ConsumerRecord<K, V>> iterator = consumer.poll(timeout).iterator();
- while (iterator.hasNext()) {
- ConsumerRecord<K, V> record = iterator.next();
- result.add(record);
- processedAtLeastOneRecord = true;
- }
-
- if (!processedAtLeastOneRecord) {
- break;
- }
- }
- consumer.commitSync();
- }
-
- return UnmodifiableList.decorate(result);
+ Properties properties, String topic) {
+ final boolean committed =
+ Objects.equals(
+ properties.getProperty(
+ ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_uncommitted"),
+ "read_committed");
+ return UnmodifiableList.decorate(
+ KafkaUtil.drainAllRecordsFromTopic(topic, properties, committed));
Review comment:
Maybe offer 2 overloads of `drainAllRecordsFromTopic` one with properties, one with URL+committed flag?
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaUtil.java
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.kafka.sink;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/** Collection of methods to interact with a Kafka cluster. */
+public class KafkaUtil {
+
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaUtil.class);
+ private static final Duration CONSUMER_POLL_DURATION = Duration.ofSeconds(1);
+
+ private KafkaUtil() {}
+
+ /**
+ * Drain all records available from the given topic from the beginning until the current highest
+ * offset.
+ *
+ * <p>This method will fetch the latest offsets for the partitions once and only return records
+ * until that point.
+ *
+ * @param topic to fetch from
+ * @param properties used to configure the created {@link KafkaConsumer}
+ * @param committed determines the mode {@link ConsumerConfig#ISOLATION_LEVEL_CONFIG} with which
+ * the consumer reads the records.
+ * @return all {@link ConsumerRecord} in the topic
+ * @throws KafkaException
+ */
+ public static List<ConsumerRecord<byte[], byte[]>> drainAllRecordsFromTopic(
+ String topic, Properties properties, boolean committed) throws KafkaException {
+ final Properties consumerConfig = new Properties();
+ consumerConfig.putAll(properties);
+ consumerConfig.put("key.deserializer", ByteArrayDeserializer.class.getName());
+ consumerConfig.put("value.deserializer", ByteArrayDeserializer.class.getName());
+ consumerConfig.put(
+ ConsumerConfig.ISOLATION_LEVEL_CONFIG,
+ committed ? "read_committed" : "read_uncommitted");
+ try (KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(consumerConfig)) {
+ Map<Integer, TopicPartition> assignments = getAllPartitions(consumer, topic);
+ Map<TopicPartition, Long> endOffsets = consumer.endOffsets(assignments.values());
+ consumer.assign(assignments.values());
+ consumer.seekToBeginning(assignments.values());
+
+ final List<ConsumerRecord<byte[], byte[]>> consumerRecords = new ArrayList<>();
+ while (!assignments.isEmpty()) {
+ consumer.assign(assignments.values());
+ ConsumerRecords<byte[], byte[]> records = consumer.poll(CONSUMER_POLL_DURATION);
+ LOG.info("Fetched {} records from topic {}.", records.count(), topic);
+
+ // Remove partitions from polling which have reached its end.
+ final Iterator<Map.Entry<Integer, TopicPartition>> assignmentIterator =
+ assignments.entrySet().iterator();
+ while (assignmentIterator.hasNext()) {
+ final Map.Entry<Integer, TopicPartition> assignment = assignmentIterator.next();
+ final TopicPartition topicPartition = assignment.getValue();
+ final long position = consumer.position(topicPartition);
+ final long endOffset = endOffsets.get(topicPartition);
+ LOG.info(
+ "Endoffset {} and current position {} for partition {}",
+ endOffset,
+ position,
+ assignment.getKey());
+ if (endOffset - position > 0) {
+ continue;
+ }
+ assignmentIterator.remove();
+ }
Review comment:
We could have a `finishedPartitions` list in each iteration. Then we don't need the Iterator as we would only modify assignments in a batch way at the end of the iteration. Further, we would only call `consumer.assign` on a change.
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaUtil.java
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.kafka.sink;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/** Collection of methods to interact with a Kafka cluster. */
+public class KafkaUtil {
+
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaUtil.class);
+ private static final Duration CONSUMER_POLL_DURATION = Duration.ofSeconds(1);
+
+ private KafkaUtil() {}
+
+ /**
+ * Drain all records available from the given topic from the beginning until the current highest
+ * offset.
+ *
+ * <p>This method will fetch the latest offsets for the partitions once and only return records
+ * until that point.
+ *
+ * @param topic to fetch from
+ * @param properties used to configure the created {@link KafkaConsumer}
+ * @param committed determines the mode {@link ConsumerConfig#ISOLATION_LEVEL_CONFIG} with which
+ * the consumer reads the records.
+ * @return all {@link ConsumerRecord} in the topic
+ * @throws KafkaException
+ */
+ public static List<ConsumerRecord<byte[], byte[]>> drainAllRecordsFromTopic(
+ String topic, Properties properties, boolean committed) throws KafkaException {
+ final Properties consumerConfig = new Properties();
+ consumerConfig.putAll(properties);
+ consumerConfig.put("key.deserializer", ByteArrayDeserializer.class.getName());
+ consumerConfig.put("value.deserializer", ByteArrayDeserializer.class.getName());
+ consumerConfig.put(
+ ConsumerConfig.ISOLATION_LEVEL_CONFIG,
+ committed ? "read_committed" : "read_uncommitted");
+ try (KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(consumerConfig)) {
+ Map<Integer, TopicPartition> assignments = getAllPartitions(consumer, topic);
+ Map<TopicPartition, Long> endOffsets = consumer.endOffsets(assignments.values());
+ consumer.assign(assignments.values());
+ consumer.seekToBeginning(assignments.values());
+
+ final List<ConsumerRecord<byte[], byte[]>> consumerRecords = new ArrayList<>();
+ while (!assignments.isEmpty()) {
+ consumer.assign(assignments.values());
+ ConsumerRecords<byte[], byte[]> records = consumer.poll(CONSUMER_POLL_DURATION);
+ LOG.info("Fetched {} records from topic {}.", records.count(), topic);
+
+ // Remove partitions from polling which have reached its end.
+ final Iterator<Map.Entry<Integer, TopicPartition>> assignmentIterator =
+ assignments.entrySet().iterator();
+ while (assignmentIterator.hasNext()) {
+ final Map.Entry<Integer, TopicPartition> assignment = assignmentIterator.next();
+ final TopicPartition topicPartition = assignment.getValue();
+ final long position = consumer.position(topicPartition);
+ final long endOffset = endOffsets.get(topicPartition);
+ LOG.info(
+ "Endoffset {} and current position {} for partition {}",
+ endOffset,
+ position,
+ assignment.getKey());
Review comment:
Is this the only place where we need the key?
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaUtil.java
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.kafka.sink;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/** Collection of methods to interact with a Kafka cluster. */
+public class KafkaUtil {
+
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaUtil.class);
+ private static final Duration CONSUMER_POLL_DURATION = Duration.ofSeconds(1);
+
+ private KafkaUtil() {}
+
+ /**
+ * Drain all records available from the given topic from the beginning until the current highest
+ * offset.
+ *
+ * <p>This method will fetch the latest offsets for the partitions once and only return records
+ * until that point.
+ *
+ * @param topic to fetch from
+ * @param properties used to configure the created {@link KafkaConsumer}
+ * @param committed determines the mode {@link ConsumerConfig#ISOLATION_LEVEL_CONFIG} with which
+ * the consumer reads the records.
+ * @return all {@link ConsumerRecord} in the topic
+ * @throws KafkaException
+ */
+ public static List<ConsumerRecord<byte[], byte[]>> drainAllRecordsFromTopic(
+ String topic, Properties properties, boolean committed) throws KafkaException {
+ final Properties consumerConfig = new Properties();
+ consumerConfig.putAll(properties);
+ consumerConfig.put("key.deserializer", ByteArrayDeserializer.class.getName());
+ consumerConfig.put("value.deserializer", ByteArrayDeserializer.class.getName());
+ consumerConfig.put(
+ ConsumerConfig.ISOLATION_LEVEL_CONFIG,
+ committed ? "read_committed" : "read_uncommitted");
+ try (KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(consumerConfig)) {
+ Map<Integer, TopicPartition> assignments = getAllPartitions(consumer, topic);
+ Map<TopicPartition, Long> endOffsets = consumer.endOffsets(assignments.values());
+ consumer.assign(assignments.values());
+ consumer.seekToBeginning(assignments.values());
+
+ final List<ConsumerRecord<byte[], byte[]>> consumerRecords = new ArrayList<>();
+ while (!assignments.isEmpty()) {
+ consumer.assign(assignments.values());
+ ConsumerRecords<byte[], byte[]> records = consumer.poll(CONSUMER_POLL_DURATION);
+ LOG.info("Fetched {} records from topic {}.", records.count(), topic);
Review comment:
debug
--
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
[GitHub] [flink] AHeise commented on a change in pull request #17406: [FLINK-24405][tests] Introduce util to reliably drain all messages from a kafka topic
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #17406:
URL: https://github.com/apache/flink/pull/17406#discussion_r722956800
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
##########
@@ -279,32 +279,16 @@ public String getVersion() {
}
@Override
+ @SuppressWarnings("unchecked")
public <K, V> Collection<ConsumerRecord<K, V>> getAllRecordsFromTopic(
- Properties properties, String topic, int partition, long timeout) {
- List<ConsumerRecord<K, V>> result = new ArrayList<>();
-
- try (KafkaConsumer<K, V> consumer = new KafkaConsumer<>(properties)) {
- consumer.assign(Arrays.asList(new TopicPartition(topic, partition)));
-
- while (true) {
- boolean processedAtLeastOneRecord = false;
-
- // wait for new records with timeout and break the loop if we didn't get any
- Iterator<ConsumerRecord<K, V>> iterator = consumer.poll(timeout).iterator();
- while (iterator.hasNext()) {
- ConsumerRecord<K, V> record = iterator.next();
- result.add(record);
- processedAtLeastOneRecord = true;
- }
-
- if (!processedAtLeastOneRecord) {
- break;
- }
- }
- consumer.commitSync();
- }
-
- return UnmodifiableList.decorate(result);
+ Properties properties, String topic) {
+ final boolean committed =
+ Objects.equals(
+ properties.getProperty(
+ ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_uncommitted"),
+ "read_committed");
+ return UnmodifiableList.decorate(
+ KafkaUtil.drainAllRecordsFromTopic(topic, properties, committed));
Review comment:
It still feels hacky to pull out committed from properties, just to reset the properties to that value. Maybe 2 overloads where committed is optional and overriddes the properties?
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #17406: [FLINK-24405][tests] Introduce util to reliably drain all messages from a kafka topic
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17406:
URL: https://github.com/apache/flink/pull/17406#issuecomment-932253575
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "21ea516945a020245a9699d6bed90e0d3c4cb2b7",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=24708",
"triggerID" : "21ea516945a020245a9699d6bed90e0d3c4cb2b7",
"triggerType" : "PUSH"
}, {
"hash" : "d861fda62f147af76f829459b0a31dcc20c2c257",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=24775",
"triggerID" : "d861fda62f147af76f829459b0a31dcc20c2c257",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 21ea516945a020245a9699d6bed90e0d3c4cb2b7 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=24708)
* d861fda62f147af76f829459b0a31dcc20c2c257 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=24775)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #17406: [FLINK-24405][tests] Introduce util to reliably drain all messages from a kafka topic
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17406:
URL: https://github.com/apache/flink/pull/17406#issuecomment-932253575
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "21ea516945a020245a9699d6bed90e0d3c4cb2b7",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=24708",
"triggerID" : "21ea516945a020245a9699d6bed90e0d3c4cb2b7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 21ea516945a020245a9699d6bed90e0d3c4cb2b7 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=24708)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #17406: [FLINK-24405][tests] Introduce util to reliably drain all messages from a kafka topic
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17406:
URL: https://github.com/apache/flink/pull/17406#issuecomment-932253575
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "21ea516945a020245a9699d6bed90e0d3c4cb2b7",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=24708",
"triggerID" : "21ea516945a020245a9699d6bed90e0d3c4cb2b7",
"triggerType" : "PUSH"
}, {
"hash" : "d861fda62f147af76f829459b0a31dcc20c2c257",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d861fda62f147af76f829459b0a31dcc20c2c257",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 21ea516945a020245a9699d6bed90e0d3c4cb2b7 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=24708)
* d861fda62f147af76f829459b0a31dcc20c2c257 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #17406: [FLINK-24405][tests] Introduce util to reliably drain all messages from a kafka topic
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17406:
URL: https://github.com/apache/flink/pull/17406#issuecomment-932253575
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "21ea516945a020245a9699d6bed90e0d3c4cb2b7",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=24708",
"triggerID" : "21ea516945a020245a9699d6bed90e0d3c4cb2b7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 21ea516945a020245a9699d6bed90e0d3c4cb2b7 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=24708)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #17406: [FLINK-24405][tests] Introduce util to reliably drain all messages from a kafka topic
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17406:
URL: https://github.com/apache/flink/pull/17406#issuecomment-932253575
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "21ea516945a020245a9699d6bed90e0d3c4cb2b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=24708",
"triggerID" : "21ea516945a020245a9699d6bed90e0d3c4cb2b7",
"triggerType" : "PUSH"
}, {
"hash" : "d861fda62f147af76f829459b0a31dcc20c2c257",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=24775",
"triggerID" : "d861fda62f147af76f829459b0a31dcc20c2c257",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d861fda62f147af76f829459b0a31dcc20c2c257 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=24775)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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