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/03/12 01:21:24 UTC
[GitHub] [flink] lindong28 opened a new pull request #15161: Flink 20114
lindong28 opened a new pull request #15161:
URL: https://github.com/apache/flink/pull/15161
## What is the purpose of the change
This PR fixed a few bugs related to `KafkaSource`. See "Brief change log" for the list of bug fixes made in this PR.
## Brief change log
Bug fixes:
- `KafkaSourceReader` should not commit offsets for partitions whose offsets have not been initialized.
- `SourceCoordinatorContext` should not log and fail job again if it receives InterruptedException after it is closed.
- `SourceOperatorStreamTask` should be closed synchronously to avoid ClassNotFoundException.
- `PartitionOffsetsRetrieverImpl.committedOffsets()` should handle the case without committed offsets.
- `SourceOperatorStreamTask` should check the committed offset first before using OffsetResetStrategy.
- Auto offset commit should be disabled by default.
- `SourceCoordinatorContext` should fail job if it fails to send event to subtasks.
Usability improvements:
- Reduce the offset commit logging verbosity from INFO to DEBUG.
- `SourceOperatorStreamTask` should report numRecordsOutCount.
- `KafkaSourceEnumerator` should close the admin client early if periodic partition discovery is disabled.
- Remove the unused `close.timeout.ms` config.
Tests:
- Add IT cases for KafkaSource by migrating IT cases from FlinkKafkaConsumer.
## Verifying this change
- Added tests in `KafkaSourceITCase`.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (no)
- If yes, how is the feature documented? (not applicable)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lindong28 commented on a change in pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #15161:
URL: https://github.com/apache/flink/pull/15161#discussion_r600214562
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/initializer/SpecifiedOffsetsInitializer.java
##########
@@ -60,6 +60,12 @@
}
}
if (!toLookup.isEmpty()) {
+ // First check the committed offsets.
Review comment:
If the committed offsets do not exist, regardless of whether group ID is null, we will either use the specified offset (if it is specified) or fallback to use the offsetResetStrategy. We will throw an exception only if we want to use the `offsetResetStrategy` and `offsetResetStrategy == NONE`.
According to the code [here](https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java#L646
), the existing `FlinkKafkaConsumer` will use the committed offset if there is no user-specified offset.
Because the existing `FlinkKafkaConsumer` goes through the steps [(1) specified (2) committed (3) fallback], I think it is desirable to stick to this behavior unless we have good use-case for the other behavior (e.g. [(1) specified (2) fallback]),
What do you think?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "52088a88b0c5666ef2522248057de9f5a787452c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14779",
"triggerID" : "52088a88b0c5666ef2522248057de9f5a787452c",
"triggerType" : "PUSH"
}, {
"hash" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"triggerType" : "PUSH"
}, {
"hash" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15120",
"triggerID" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"triggerType" : "PUSH"
}, {
"hash" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15197",
"triggerID" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"triggerType" : "PUSH"
}, {
"hash" : "3bbb86b12a43044ce3f876401c98f28c87858fbe",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15366",
"triggerID" : "3bbb86b12a43044ce3f876401c98f28c87858fbe",
"triggerType" : "PUSH"
}, {
"hash" : "c58bb5faec3f07e0b66d0e042e3e9c6082fb8e9c",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15416",
"triggerID" : "c58bb5faec3f07e0b66d0e042e3e9c6082fb8e9c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d3f59e302b9860d49ce79252aeb8bd1decaeeabf UNKNOWN
* 3bbb86b12a43044ce3f876401c98f28c87858fbe Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15366)
* c58bb5faec3f07e0b66d0e042e3e9c6082fb8e9c Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15416)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "52088a88b0c5666ef2522248057de9f5a787452c",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14779",
"triggerID" : "52088a88b0c5666ef2522248057de9f5a787452c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6b317a312546abfb536c1358a331f1878e65e5d3 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754) Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692)
* 52088a88b0c5666ef2522248057de9f5a787452c Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14779)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c45b554fee8aba36169cffe7b2da6621e7beb9d3 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] becketqin commented on a change in pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
becketqin commented on a change in pull request #15161:
URL: https://github.com/apache/flink/pull/15161#discussion_r600612782
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/RecreateOnResetOperatorCoordinator.java
##########
@@ -74,7 +74,8 @@ public void start() throws Exception {
@Override
public void close() throws Exception {
closed = true;
- coordinator.closeAsync(closingTimeoutMs);
+ // Wait for coordinator close before destructing any user class loader.
+ coordinator.closeAsync(closingTimeoutMs).get();
Review comment:
I do not remember the exact original issue I saw there. It could be that the `SplitEnumerator` is still doing something (e.g. partition discovery) before it is closed.
I am not sure if we need to worry about this here. For two reasons:
1. In Flink the closure of all the components has always been synchronous, so closing the `OperatorCoordinator` here synchronously seems fine. This call may only block if the `SplitEnumerator` itself blocks on closure. Although this might be something we cannot not avoid if there is user code, it is very likely the synchronous closure is still non-blocking. For example, `CheckpointCoordinator.shutdown()` may block on the closure of master hooks.
2. It is the JM's responsibility to ensure the components are closed normally to avoid potential resource leak. Theoretically speaking, we can make all the component closure asynchronous so the main thread of JM is totally non-blocking, but chances are it still has to wait until all the components are closed normally before it can move to the next step. So to some extent, blocking on `closure` is a correctness requirement.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "52088a88b0c5666ef2522248057de9f5a787452c",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14779",
"triggerID" : "52088a88b0c5666ef2522248057de9f5a787452c",
"triggerType" : "PUSH"
}, {
"hash" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"triggerType" : "PUSH"
}, {
"hash" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 52088a88b0c5666ef2522248057de9f5a787452c Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14779)
* d3f59e302b9860d49ce79252aeb8bd1decaeeabf UNKNOWN
* 728e9c6d1562fc250f942eaf147a3aacae95c436 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797170126
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 6c17ec7241323224d3053134e82353c63f28a24f (Sat Aug 28 10:34:46 UTC 2021)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
* **This pull request references an unassigned [Jira ticket](https://issues.apache.org/jira/browse/FLINK-20114).** According to the [code contribution guide](https://flink.apache.org/contributing/contribute-code.html), tickets need to be assigned before starting with the implementation work.
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ✅ 1. The [description] looks good.
- Approved by @lindong28
* ❓ 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] becketqin merged pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
becketqin merged pull request #15161:
URL: https://github.com/apache/flink/pull/15161
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
} ]
}-->
## CI report:
* 6b317a312546abfb536c1358a331f1878e65e5d3 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754) Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "52088a88b0c5666ef2522248057de9f5a787452c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "52088a88b0c5666ef2522248057de9f5a787452c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6b317a312546abfb536c1358a331f1878e65e5d3 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754) Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692)
* 52088a88b0c5666ef2522248057de9f5a787452c 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 10afa89b37842487ecbcb3e4835d7919b35406c0 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "52088a88b0c5666ef2522248057de9f5a787452c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14779",
"triggerID" : "52088a88b0c5666ef2522248057de9f5a787452c",
"triggerType" : "PUSH"
}, {
"hash" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"triggerType" : "PUSH"
}, {
"hash" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15120",
"triggerID" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"triggerType" : "PUSH"
}, {
"hash" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15197",
"triggerID" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"triggerType" : "PUSH"
}, {
"hash" : "3bbb86b12a43044ce3f876401c98f28c87858fbe",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15366",
"triggerID" : "3bbb86b12a43044ce3f876401c98f28c87858fbe",
"triggerType" : "PUSH"
}, {
"hash" : "c58bb5faec3f07e0b66d0e042e3e9c6082fb8e9c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15416",
"triggerID" : "c58bb5faec3f07e0b66d0e042e3e9c6082fb8e9c",
"triggerType" : "PUSH"
}, {
"hash" : "6fc6cc9a2e6c26e9e8fa62b481237231394078eb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15507",
"triggerID" : "6fc6cc9a2e6c26e9e8fa62b481237231394078eb",
"triggerType" : "PUSH"
}, {
"hash" : "6c17ec7241323224d3053134e82353c63f28a24f",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15649",
"triggerID" : "6c17ec7241323224d3053134e82353c63f28a24f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d3f59e302b9860d49ce79252aeb8bd1decaeeabf UNKNOWN
* 6c17ec7241323224d3053134e82353c63f28a24f Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15649)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] StephanEwen commented on a change in pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
StephanEwen commented on a change in pull request #15161:
URL: https://github.com/apache/flink/pull/15161#discussion_r599670918
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContext.java
##########
@@ -267,6 +270,12 @@ void failJob(Throwable cause) {
}
void handleUncaughtExceptionFromAsyncCall(Throwable t) {
+ if (closed) {
+ // We do not trigger a job failure again if the source coordinator is already closed.
+ LOG.debug("Caught exception when shutting down the SourceCoordinatorContext.", t);
Review comment:
Users may still be confused by this message here, showing an Exception and a stack trace.
I would suggest to not even log here, because the contract is that exceptions after closing are suppressed. We do a similar thing during task cancellation: as soon as the status is CANCELLED, all exceptions are suppressed. It gives the cleanest logs and user experience.
Alternatively, if you think logging here is crucial, I would change the log message to something like: "Ignoring async exception after closing the coordinator." so that users know they do not need to worry about that exception.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContext.java
##########
@@ -103,18 +106,22 @@ public SourceCoordinatorContext(
SimpleVersionedSerializer<SplitT> splitSerializer) {
this(
coordinatorExecutor,
+ Executors.newScheduledThreadPool(
+ numWorkerThreads,
+ new ExecutorThreadFactory(
+ coordinatorThreadFactory.getCoordinatorThreadName() + "-worker")),
coordinatorThreadFactory,
- numWorkerThreads,
operatorCoordinatorContext,
splitSerializer,
new SplitAssignmentTracker<>());
}
// Package private method for unit test.
+ @VisibleForTesting
Review comment:
Side note: I think this is the right pattern used here.
In the majority of cases, classes should have one core constructor that accepts all the internal components (dependencies) as arguments (except maybe simple utility types, like lists and maps that are initially empty). That constructor is needed to inject special dependencies for testing (like mocks).
Then there are often utility constructors or factory methods so that not everyone who instantiates the class needs to instantiate all dependencies as well.
I would encourage to try and apply that pattern as much as possible, it is quite useful in the long run.
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContextTest.java
##########
@@ -182,6 +191,44 @@ public void testSnapshotAndRestore() throws Exception {
restoredTracker.assignmentsByCheckpointId());
}
+ @Test
+ public void testCallableInterruptedDuringShutdownDoNotFailJob() throws InterruptedException {
+ AtomicReference<Throwable> expectedError = new AtomicReference<>(null);
+
+ ManuallyTriggeredScheduledExecutorService manualWorkerExecutor =
+ new ManuallyTriggeredScheduledExecutorService();
+ ManuallyTriggeredScheduledExecutorService manualCoordinatorExecutor =
+ new ManuallyTriggeredScheduledExecutorService();
+
+ SourceCoordinatorContext<MockSourceSplit> testingContext =
+ new SourceCoordinatorContext<>(
+ manualCoordinatorExecutor,
+ manualWorkerExecutor,
+ new SourceCoordinatorProvider.CoordinatorExecutorThreadFactory(
+ TEST_OPERATOR_ID.toHexString(), getClass().getClassLoader()),
+ operatorCoordinatorContext,
+ new MockSourceSplitSerializer(),
+ splitSplitAssignmentTracker);
+
+ testingContext.callAsync(
+ () -> {
+ throw new InterruptedException();
+ },
+ (ignored, e) -> {
+ if (e != null) {
+ expectedError.set(e);
+ throw new RuntimeException(e);
+ }
+ });
+
+ manualWorkerExecutor.triggerAll();
Review comment:
Is this needed, or "just in case"?
(it looks like the `callAsync(...)` calls go directly to worker executor).
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/initializer/SpecifiedOffsetsInitializer.java
##########
@@ -60,6 +60,12 @@
}
}
if (!toLookup.isEmpty()) {
+ // First check the committed offsets.
Review comment:
What actually happens here when committed offsets do not exist, because the user didn't specify a group ID?
I assume then the group ID is null, will the PartitionOffsetsRetriever throw an exception in that case?
Is it desirable that we always go through the steps [(1) specified (2) committed (3) fallback], or is it a separate case and the previous case [(1) specified (2) fallback] is equally valid?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java
##########
@@ -88,11 +90,15 @@ public void init() throws Exception {
input = new StreamTaskSourceInput<>(sourceOperator, 0, 0);
}
+ CountingOutput<T> countingOutput =
Review comment:
How hard is it to add the counting to the `AsyncDataOutputToOutput` class instead, instead of adding the `CountingOutput` as a wrapper?
This path is the most performance critical path in the system, the pushing of events through the chain. Any wrapping here has the potential of adding performance overhead, through virtual method calls. The exact effect is hard to estimate, because it depends in the end on how well the JIT can profile and optimize the whole thing.
From everything I understand about the commonly used JIT, the `Output` interface and its subclasses are already a tricky case, because it has so many implementations that are used and mixed at the same time. We are more JIT friendly when we don't add more subclasses (and I think we should in fact try to reduce the subclasses that exist already).
This may be a very small bit individually, but these small bits tend to add up in the long run as well, so always a good idea to continuously keep them in mind.
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java
##########
@@ -435,6 +435,7 @@ private void parseAndSetRequiredProperties() {
true);
maybeOverride(
ConsumerConfig.GROUP_ID_CONFIG, "KafkaSource-" + new Random().nextLong(), false);
+ maybeOverride(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false", false);
Review comment:
I think the old Kafka Source always set this to `false` and did not allow an override.
Is there ever a case when we want "auto offset commit" to be used? Committing on checkpoint completion is more meaningful, I think, when users want offsets in Kafka.
I am not opposed to allowing this, I am just curious if this can lead to surprises by users who do not understand well that Flink manages the offsets directly.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/RecreateOnResetOperatorCoordinator.java
##########
@@ -74,7 +74,8 @@ public void start() throws Exception {
@Override
public void close() throws Exception {
closed = true;
- coordinator.closeAsync(closingTimeoutMs);
+ // Wait for coordinator close before destructing any user class loader.
+ coordinator.closeAsync(closingTimeoutMs).get();
Review comment:
I don't think we can do this here. All methods form the coordinator must be non-blocking, because they are called by the JobManager's scheduler thread. And that thread must not block, otherwise the whole job may become unresponsive.
I'd like to understand a bit better how this is necessary to avoid `ClassNotFoundException`. Where is that exception thrown and can we prevent this differently, similar to the case where exceptions from callbacks are ignored once the SourceEnumeratorContext is closed.
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java
##########
@@ -442,10 +442,8 @@ private void parseAndSetRequiredProperties() {
true);
// If the source is bounded, do not run periodic partition discovery.
- if (maybeOverride(
- KafkaSourceOptions.PARTITION_DISCOVERY_INTERVAL_MS.key(),
- "-1",
- boundedness == Boundedness.BOUNDED)) {
+ if (boundedness == Boundedness.BOUNDED &&
Review comment:
I don't quite agree with this. Is there a ticket that describes the reasoning behind this setting?
Partition discovery is good to support, but I wouldn't activate it by default. The reasons are that
1. This is a surprising change form the behavior of the previous Kafka Source
2. I have met quite a few users that appreciate the "stable" subscription as a default model. Additional partitions suddenly appearing is a challenge in many setups anyways, and also frequently messes with order guarantees.
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java
##########
@@ -407,7 +407,12 @@ public PartitionOffsetsRetrieverImpl(
.thenApply(
result -> {
Map<TopicPartition, Long> offsets = new HashMap<>();
- result.forEach((tp, oam) -> offsets.put(tp, oam.offset()));
+ result.forEach(
Review comment:
Is it reasonable to guard this with a unit test, or is that caught as part of some other tests (like broader tests) already?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "52088a88b0c5666ef2522248057de9f5a787452c",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14779",
"triggerID" : "52088a88b0c5666ef2522248057de9f5a787452c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 52088a88b0c5666ef2522248057de9f5a787452c Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14779)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "52088a88b0c5666ef2522248057de9f5a787452c",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14779",
"triggerID" : "52088a88b0c5666ef2522248057de9f5a787452c",
"triggerType" : "PUSH"
}, {
"hash" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"triggerType" : "PUSH"
}, {
"hash" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15120",
"triggerID" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 52088a88b0c5666ef2522248057de9f5a787452c Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14779)
* d3f59e302b9860d49ce79252aeb8bd1decaeeabf UNKNOWN
* 728e9c6d1562fc250f942eaf147a3aacae95c436 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15120)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lindong28 commented on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
lindong28 commented on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-805642589
@StephanEwen Thank you for the detailed review! The PR's code has been updated according to the discussion.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] becketqin commented on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
becketqin commented on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-809801052
Merged to master.
8d2fa1bb81eff790bdc21a69d15a17aa074010ae
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lindong28 commented on a change in pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #15161:
URL: https://github.com/apache/flink/pull/15161#discussion_r601981570
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamMultipleInputProcessorFactory.java
##########
@@ -351,7 +351,7 @@ public StreamTaskSourceOutput(
WatermarkGauge inputWatermarkGauge,
MultiStreamStreamStatusTracker streamStatusTracker,
int inputIndex) {
- super(chainedSourceOutput, streamStatusMaintainer, inputWatermarkGauge);
+ super(chainedSourceOutput, streamStatusMaintainer, inputWatermarkGauge, null);
Review comment:
Sounds good. It is updated as suggested.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "52088a88b0c5666ef2522248057de9f5a787452c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14779",
"triggerID" : "52088a88b0c5666ef2522248057de9f5a787452c",
"triggerType" : "PUSH"
}, {
"hash" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"triggerType" : "PUSH"
}, {
"hash" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15120",
"triggerID" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"triggerType" : "PUSH"
}, {
"hash" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15197",
"triggerID" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"triggerType" : "PUSH"
}, {
"hash" : "3bbb86b12a43044ce3f876401c98f28c87858fbe",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15366",
"triggerID" : "3bbb86b12a43044ce3f876401c98f28c87858fbe",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d3f59e302b9860d49ce79252aeb8bd1decaeeabf UNKNOWN
* 3bbb86b12a43044ce3f876401c98f28c87858fbe Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15366)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] StephanEwen commented on a change in pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
StephanEwen commented on a change in pull request #15161:
URL: https://github.com/apache/flink/pull/15161#discussion_r600300881
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/initializer/SpecifiedOffsetsInitializer.java
##########
@@ -60,6 +60,12 @@
}
}
if (!toLookup.isEmpty()) {
+ // First check the committed offsets.
Review comment:
I think it sounds good, I just wanted to understand the reasoning.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lindong28 commented on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
lindong28 commented on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-799917097
@flinkbot run azure
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lindong28 commented on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
lindong28 commented on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-806310235
@StephanEwen @becketqin Thanks for the review! I think all issues are resolved. Could you take another look?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lindong28 commented on a change in pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #15161:
URL: https://github.com/apache/flink/pull/15161#discussion_r600109069
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContext.java
##########
@@ -267,6 +270,12 @@ void failJob(Throwable cause) {
}
void handleUncaughtExceptionFromAsyncCall(Throwable t) {
+ if (closed) {
+ // We do not trigger a job failure again if the source coordinator is already closed.
+ LOG.debug("Caught exception when shutting down the SourceCoordinatorContext.", t);
Review comment:
Thank you Stephan for the explanation. Sounds good. I have removed this logging.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617)
* c45b554fee8aba36169cffe7b2da6621e7beb9d3 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c45b554fee8aba36169cffe7b2da6621e7beb9d3 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661)
* 6b317a312546abfb536c1358a331f1878e65e5d3 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 10afa89b37842487ecbcb3e4835d7919b35406c0 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "52088a88b0c5666ef2522248057de9f5a787452c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14779",
"triggerID" : "52088a88b0c5666ef2522248057de9f5a787452c",
"triggerType" : "PUSH"
}, {
"hash" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"triggerType" : "PUSH"
}, {
"hash" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15120",
"triggerID" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"triggerType" : "PUSH"
}, {
"hash" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d3f59e302b9860d49ce79252aeb8bd1decaeeabf UNKNOWN
* 728e9c6d1562fc250f942eaf147a3aacae95c436 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15120)
* 871e9d892b6da0907cc3c5a1d811f145b0bb0f47 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lindong28 commented on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
lindong28 commented on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797173893
@flinkbot approve description
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] StephanEwen commented on a change in pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
StephanEwen commented on a change in pull request #15161:
URL: https://github.com/apache/flink/pull/15161#discussion_r600299391
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java
##########
@@ -435,6 +435,7 @@ private void parseAndSetRequiredProperties() {
true);
maybeOverride(
ConsumerConfig.GROUP_ID_CONFIG, "KafkaSource-" + new Random().nextLong(), false);
+ maybeOverride(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false", false);
Review comment:
I guess for jobs that do not checkpoint, it makes sense yes. I am not sure anyone runs Flink without checkpoints, but fair enough.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lindong28 commented on a change in pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #15161:
URL: https://github.com/apache/flink/pull/15161#discussion_r600110083
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContext.java
##########
@@ -103,18 +106,22 @@ public SourceCoordinatorContext(
SimpleVersionedSerializer<SplitT> splitSerializer) {
this(
coordinatorExecutor,
+ Executors.newScheduledThreadPool(
+ numWorkerThreads,
+ new ExecutorThreadFactory(
+ coordinatorThreadFactory.getCoordinatorThreadName() + "-worker")),
coordinatorThreadFactory,
- numWorkerThreads,
operatorCoordinatorContext,
splitSerializer,
new SplitAssignmentTracker<>());
}
// Package private method for unit test.
+ @VisibleForTesting
Review comment:
Sounds good. Will continue to follow this pattern in the future.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
} ]
}-->
## CI report:
* 6b317a312546abfb536c1358a331f1878e65e5d3 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692) Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617)
* c45b554fee8aba36169cffe7b2da6621e7beb9d3 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "52088a88b0c5666ef2522248057de9f5a787452c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14779",
"triggerID" : "52088a88b0c5666ef2522248057de9f5a787452c",
"triggerType" : "PUSH"
}, {
"hash" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"triggerType" : "PUSH"
}, {
"hash" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15120",
"triggerID" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"triggerType" : "PUSH"
}, {
"hash" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15197",
"triggerID" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"triggerType" : "PUSH"
}, {
"hash" : "3bbb86b12a43044ce3f876401c98f28c87858fbe",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15366",
"triggerID" : "3bbb86b12a43044ce3f876401c98f28c87858fbe",
"triggerType" : "PUSH"
}, {
"hash" : "c58bb5faec3f07e0b66d0e042e3e9c6082fb8e9c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15416",
"triggerID" : "c58bb5faec3f07e0b66d0e042e3e9c6082fb8e9c",
"triggerType" : "PUSH"
}, {
"hash" : "6fc6cc9a2e6c26e9e8fa62b481237231394078eb",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15507",
"triggerID" : "6fc6cc9a2e6c26e9e8fa62b481237231394078eb",
"triggerType" : "PUSH"
}, {
"hash" : "6c17ec7241323224d3053134e82353c63f28a24f",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15649",
"triggerID" : "6c17ec7241323224d3053134e82353c63f28a24f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d3f59e302b9860d49ce79252aeb8bd1decaeeabf UNKNOWN
* 6fc6cc9a2e6c26e9e8fa62b481237231394078eb Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15507)
* 6c17ec7241323224d3053134e82353c63f28a24f Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15649)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 10afa89b37842487ecbcb3e4835d7919b35406c0 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508)
* 2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] becketqin commented on a change in pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
becketqin commented on a change in pull request #15161:
URL: https://github.com/apache/flink/pull/15161#discussion_r600633809
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java
##########
@@ -442,10 +442,8 @@ private void parseAndSetRequiredProperties() {
true);
// If the source is bounded, do not run periodic partition discovery.
- if (maybeOverride(
- KafkaSourceOptions.PARTITION_DISCOVERY_INTERVAL_MS.key(),
- "-1",
- boundedness == Boundedness.BOUNDED)) {
+ if (boundedness == Boundedness.BOUNDED &&
Review comment:
I think it is fine either way if there are customers expecting the partitions to not change. I turned the partition discovery on by default because there are also customers got surprised that the newly added partitions are not picked up automatically.
One thing that might be worth clarifying is that keeping reading from a static partition set cannot guarantee the message orders in case of Kafka partition expansion.
For example, assuming there are 2 partition in the beginning. And a key of hash `H` goes to partition `H%2`. A key whose hash value is 3 may move from partition 1 to partition 0 after a partition expansion from 2 partitions to 3 partitions. So even if the Flink job keeps consuming from partition 0 and 1 without picking up the new partitions, the message order is broken.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lindong28 commented on a change in pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #15161:
URL: https://github.com/apache/flink/pull/15161#discussion_r600205574
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java
##########
@@ -435,6 +435,7 @@ private void parseAndSetRequiredProperties() {
true);
maybeOverride(
ConsumerConfig.GROUP_ID_CONFIG, "KafkaSource-" + new Random().nextLong(), false);
+ maybeOverride(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false", false);
Review comment:
Hmm... from what I can tell, old Kafka Source (i.e. `FlinkKafkaConsumer`) actually allows `enable.auto.commit` to be set to `true`. This is because the `FlinkKafkaConsumer` [constructor](https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer.java#L210) allows users to specify `enable.auto.commit` in `Properties` which is used to instantiated Kafka consumer.
And it is mentioned in [here](https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer.java#L251) that we will overwrite the "enable.auto.commit" in the properties if our offset commit mode is ON_CHECKPOINTS. This suggests that we allow `enable.auto.commit` to be true.
From what I learned from @becketqin, the use-case here is that, by enabling auto offset commit (when the Flink job does not do checkpoint), we allow third-party Kafka tools to track the progress of the Kafka consumer.
Does this make sense?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "52088a88b0c5666ef2522248057de9f5a787452c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14779",
"triggerID" : "52088a88b0c5666ef2522248057de9f5a787452c",
"triggerType" : "PUSH"
}, {
"hash" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"triggerType" : "PUSH"
}, {
"hash" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15120",
"triggerID" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"triggerType" : "PUSH"
}, {
"hash" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15197",
"triggerID" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"triggerType" : "PUSH"
}, {
"hash" : "3bbb86b12a43044ce3f876401c98f28c87858fbe",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15366",
"triggerID" : "3bbb86b12a43044ce3f876401c98f28c87858fbe",
"triggerType" : "PUSH"
}, {
"hash" : "c58bb5faec3f07e0b66d0e042e3e9c6082fb8e9c",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15416",
"triggerID" : "c58bb5faec3f07e0b66d0e042e3e9c6082fb8e9c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d3f59e302b9860d49ce79252aeb8bd1decaeeabf UNKNOWN
* c58bb5faec3f07e0b66d0e042e3e9c6082fb8e9c Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15416)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "52088a88b0c5666ef2522248057de9f5a787452c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14779",
"triggerID" : "52088a88b0c5666ef2522248057de9f5a787452c",
"triggerType" : "PUSH"
}, {
"hash" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"triggerType" : "PUSH"
}, {
"hash" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15120",
"triggerID" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"triggerType" : "PUSH"
}, {
"hash" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15197",
"triggerID" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"triggerType" : "PUSH"
}, {
"hash" : "3bbb86b12a43044ce3f876401c98f28c87858fbe",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15366",
"triggerID" : "3bbb86b12a43044ce3f876401c98f28c87858fbe",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d3f59e302b9860d49ce79252aeb8bd1decaeeabf UNKNOWN
* 871e9d892b6da0907cc3c5a1d811f145b0bb0f47 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15197)
* 3bbb86b12a43044ce3f876401c98f28c87858fbe Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15366)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lindong28 commented on a change in pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #15161:
URL: https://github.com/apache/flink/pull/15161#discussion_r600205574
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java
##########
@@ -435,6 +435,7 @@ private void parseAndSetRequiredProperties() {
true);
maybeOverride(
ConsumerConfig.GROUP_ID_CONFIG, "KafkaSource-" + new Random().nextLong(), false);
+ maybeOverride(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false", false);
Review comment:
Hmm... from what I can tell, old Kafka Source (i.e. `FlinkKafkaConsumer`) actually allows `enable.auto.commit` to be set to `true`. This is because the `FlinkKafkaConsumer` [constructor](https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer.java#L210) allows users to specify `enable.auto.commit` in `Properties` which is used to instantiated Kafka consumer.
And it is mentioned in [here](https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer.java#L251) that we will overwrite the "enable.auto.commit" in the properties if our offset commit mode is ON_CHECKPOINTS.
From what I learned from @becketqin, the use-case here is that, by enabling auto offset commit (when the Flink job does not do checkpoint), we allow third-party Kafka tools to track the progress of the Kafka consumer.
Does this make sense?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lindong28 commented on pull request #15161: Flink 20114
Posted by GitBox <gi...@apache.org>.
lindong28 commented on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797172844
@StephanEwen @becketqin Could you help review this PR when you get time?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lindong28 commented on a change in pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #15161:
URL: https://github.com/apache/flink/pull/15161#discussion_r600297493
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java
##########
@@ -88,11 +90,15 @@ public void init() throws Exception {
input = new StreamTaskSourceInput<>(sourceOperator, 0, 0);
}
+ CountingOutput<T> countingOutput =
Review comment:
Thank you Stephan for the explanation. Yes I agree we should try to minimize the number of wrappers on the performance critical path. I have updated the PR to move `counter` logic into `AsyncDataOutputToOutput`.
I noticed that [StreamTaskSourceOutput](https://github.com/apache/flink/blob/master/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamMultipleInputProcessorFactory.java#L225) currently does not take any metric whereas [StreamTaskNetworkOutput](https://github.com/apache/flink/blob/master/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamMultipleInputProcessorFactory.java#L205), whose usage is in the same loop body as `StreamTaskSourceOutput`, takes `mainOperatorRecordsIn` as input.
I tried to find the answer in the code. Still it is not clear to me whether `StreamTaskSourceOutput` should take metrics (e.g. numRecordsIn, numRecordsOut) as input and update them. I will keep the `StreamTaskSourceOutput` as is.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #15161: Flink 20114
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797170126
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 10afa89b37842487ecbcb3e4835d7919b35406c0 (Fri Mar 12 01:31:09 UTC 2021)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
* **Invalid pull request title: No valid Jira ID provided**
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "52088a88b0c5666ef2522248057de9f5a787452c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14779",
"triggerID" : "52088a88b0c5666ef2522248057de9f5a787452c",
"triggerType" : "PUSH"
}, {
"hash" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"triggerType" : "PUSH"
}, {
"hash" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15120",
"triggerID" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"triggerType" : "PUSH"
}, {
"hash" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15197",
"triggerID" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"triggerType" : "PUSH"
}, {
"hash" : "3bbb86b12a43044ce3f876401c98f28c87858fbe",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15366",
"triggerID" : "3bbb86b12a43044ce3f876401c98f28c87858fbe",
"triggerType" : "PUSH"
}, {
"hash" : "c58bb5faec3f07e0b66d0e042e3e9c6082fb8e9c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15416",
"triggerID" : "c58bb5faec3f07e0b66d0e042e3e9c6082fb8e9c",
"triggerType" : "PUSH"
}, {
"hash" : "6fc6cc9a2e6c26e9e8fa62b481237231394078eb",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15507",
"triggerID" : "6fc6cc9a2e6c26e9e8fa62b481237231394078eb",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d3f59e302b9860d49ce79252aeb8bd1decaeeabf UNKNOWN
* 6fc6cc9a2e6c26e9e8fa62b481237231394078eb Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15507)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "52088a88b0c5666ef2522248057de9f5a787452c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14779",
"triggerID" : "52088a88b0c5666ef2522248057de9f5a787452c",
"triggerType" : "PUSH"
}, {
"hash" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"triggerType" : "PUSH"
}, {
"hash" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15120",
"triggerID" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"triggerType" : "PUSH"
}, {
"hash" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15197",
"triggerID" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"triggerType" : "PUSH"
}, {
"hash" : "3bbb86b12a43044ce3f876401c98f28c87858fbe",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15366",
"triggerID" : "3bbb86b12a43044ce3f876401c98f28c87858fbe",
"triggerType" : "PUSH"
}, {
"hash" : "c58bb5faec3f07e0b66d0e042e3e9c6082fb8e9c",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15416",
"triggerID" : "c58bb5faec3f07e0b66d0e042e3e9c6082fb8e9c",
"triggerType" : "PUSH"
}, {
"hash" : "6fc6cc9a2e6c26e9e8fa62b481237231394078eb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "6fc6cc9a2e6c26e9e8fa62b481237231394078eb",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d3f59e302b9860d49ce79252aeb8bd1decaeeabf UNKNOWN
* c58bb5faec3f07e0b66d0e042e3e9c6082fb8e9c Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15416)
* 6fc6cc9a2e6c26e9e8fa62b481237231394078eb 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 10afa89b37842487ecbcb3e4835d7919b35406c0 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508)
* 2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "52088a88b0c5666ef2522248057de9f5a787452c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14779",
"triggerID" : "52088a88b0c5666ef2522248057de9f5a787452c",
"triggerType" : "PUSH"
}, {
"hash" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"triggerType" : "PUSH"
}, {
"hash" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15120",
"triggerID" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d3f59e302b9860d49ce79252aeb8bd1decaeeabf UNKNOWN
* 728e9c6d1562fc250f942eaf147a3aacae95c436 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15120)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lindong28 commented on a change in pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #15161:
URL: https://github.com/apache/flink/pull/15161#discussion_r600198354
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java
##########
@@ -407,7 +407,12 @@ public PartitionOffsetsRetrieverImpl(
.thenApply(
result -> {
Map<TopicPartition, Long> offsets = new HashMap<>();
- result.forEach((tp, oam) -> offsets.put(tp, oam.offset()));
+ result.forEach(
Review comment:
Yes, this test is change is verified by `OffsetsInitializerTest#testSpecificOffsetsInitializer`, which is updated by this PR.
The test would call `committedOffsets(partitions)` where `partitions` include a partition that does not have committed offsets.
I verified that the test `testSpecificOffsetsInitializer` would fail if we don't update the `committedOffsets(...)` method here.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "52088a88b0c5666ef2522248057de9f5a787452c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14779",
"triggerID" : "52088a88b0c5666ef2522248057de9f5a787452c",
"triggerType" : "PUSH"
}, {
"hash" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"triggerType" : "PUSH"
}, {
"hash" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15120",
"triggerID" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"triggerType" : "PUSH"
}, {
"hash" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15197",
"triggerID" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d3f59e302b9860d49ce79252aeb8bd1decaeeabf UNKNOWN
* 871e9d892b6da0907cc3c5a1d811f145b0bb0f47 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15197)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lindong28 commented on pull request #15161: Flink 20114
Posted by GitBox <gi...@apache.org>.
lindong28 commented on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797172645
Most code of this PR comes from https://github.com/apache/flink/pull/14239 by @becketqin. I have taken effort to rebase it onto HEAD, verify the related fixes, and made a few more improvements (e.g bug fixes) related to KafkaSource.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lindong28 commented on a change in pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #15161:
URL: https://github.com/apache/flink/pull/15161#discussion_r600232561
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/RecreateOnResetOperatorCoordinator.java
##########
@@ -74,7 +74,8 @@ public void start() throws Exception {
@Override
public void close() throws Exception {
closed = true;
- coordinator.closeAsync(closingTimeoutMs);
+ // Wait for coordinator close before destructing any user class loader.
+ coordinator.closeAsync(closingTimeoutMs).get();
Review comment:
That is the only commit which I copied from https://github.com/apache/flink/pull/14239 without actually reproducing the bug. Previously @becketqin explained to me why it can avoid `ClassNotFoundException` and why it is OK to block here.
@becketqin could you help answer the questions here?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] StephanEwen commented on a change in pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
StephanEwen commented on a change in pull request #15161:
URL: https://github.com/apache/flink/pull/15161#discussion_r601620880
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamMultipleInputProcessorFactory.java
##########
@@ -351,7 +351,7 @@ public StreamTaskSourceOutput(
WatermarkGauge inputWatermarkGauge,
MultiStreamStreamStatusTracker streamStatusTracker,
int inputIndex) {
- super(chainedSourceOutput, streamStatusMaintainer, inputWatermarkGauge);
+ super(chainedSourceOutput, streamStatusMaintainer, inputWatermarkGauge, null);
Review comment:
Can we just pass a `new SimpleCounter()` here? I think that this code needs to be adjusted as well in the future to do proper record counting (register the proper counter), so we can directly operate under the assumption that there will be a counter passed in here.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java
##########
@@ -145,19 +155,25 @@ private void triggerCheckpointForExternallyInducedSource(long checkpointId) {
private final Output<StreamRecord<T>> output;
@Nullable private final WatermarkGauge inputWatermarkGauge;
+ @Nullable private final Counter numRecordsOut;
public AsyncDataOutputToOutput(
Output<StreamRecord<T>> output,
StreamStatusMaintainer streamStatusMaintainer,
- @Nullable WatermarkGauge inputWatermarkGauge) {
+ @Nullable WatermarkGauge inputWatermarkGauge,
+ @Nullable Counter numRecordsOut) {
super(streamStatusMaintainer);
this.output = checkNotNull(output);
this.inputWatermarkGauge = inputWatermarkGauge;
+ this.numRecordsOut = numRecordsOut;
}
@Override
public void emitRecord(StreamRecord<T> streamRecord) {
+ if (numRecordsOut != null) {
Review comment:
We could make the field non-nullable and skip the branch here. See comment above also.
Again, not sure if it makes a big difference, JIT probably profiles this away, but it also means one branch less for the JIT to worry about, and non-nullability is also easier for the next maintainer to reason about. Especially if the counter is going to be non-null anyways.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lindong28 commented on a change in pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #15161:
URL: https://github.com/apache/flink/pull/15161#discussion_r600993395
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/RecreateOnResetOperatorCoordinator.java
##########
@@ -74,7 +74,8 @@ public void start() throws Exception {
@Override
public void close() throws Exception {
closed = true;
- coordinator.closeAsync(closingTimeoutMs);
+ // Wait for coordinator close before destructing any user class loader.
+ coordinator.closeAsync(closingTimeoutMs).get();
Review comment:
@becketqin Thanks for the explanation!
It looks like we don't have clear idea on the root cause of `ClassNotFoundException`. And there is potential risk (probably unlikely) with making the call synchronous when user code has blocking operation.
Regarding the 2nd point above, it looks like there could be tradeoff between resource leak and blocking JobManager's scheduler thread. I suppose it is better to understand the resource leak before making JobManager's thread blocking.
And from https://issues.apache.org/jira/browse/FLINK-20114, it looks like the ClassNotFoundException is emitted in the WARNING without disrupting user's job. It seems OK to still emit this ClassNotFoundException in the log until we resolve the concerns described above.
@StephanEwen @becketqin To place on the safe side, I have removed this change from this 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "52088a88b0c5666ef2522248057de9f5a787452c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14779",
"triggerID" : "52088a88b0c5666ef2522248057de9f5a787452c",
"triggerType" : "PUSH"
}, {
"hash" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"triggerType" : "PUSH"
}, {
"hash" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15120",
"triggerID" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"triggerType" : "PUSH"
}, {
"hash" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15197",
"triggerID" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"triggerType" : "PUSH"
}, {
"hash" : "3bbb86b12a43044ce3f876401c98f28c87858fbe",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15366",
"triggerID" : "3bbb86b12a43044ce3f876401c98f28c87858fbe",
"triggerType" : "PUSH"
}, {
"hash" : "c58bb5faec3f07e0b66d0e042e3e9c6082fb8e9c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "c58bb5faec3f07e0b66d0e042e3e9c6082fb8e9c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d3f59e302b9860d49ce79252aeb8bd1decaeeabf UNKNOWN
* 3bbb86b12a43044ce3f876401c98f28c87858fbe Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15366)
* c58bb5faec3f07e0b66d0e042e3e9c6082fb8e9c 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lindong28 commented on a change in pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #15161:
URL: https://github.com/apache/flink/pull/15161#discussion_r600228593
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java
##########
@@ -442,10 +442,8 @@ private void parseAndSetRequiredProperties() {
true);
// If the source is bounded, do not run periodic partition discovery.
- if (maybeOverride(
- KafkaSourceOptions.PARTITION_DISCOVERY_INTERVAL_MS.key(),
- "-1",
- boundedness == Boundedness.BOUNDED)) {
+ if (boundedness == Boundedness.BOUNDED &&
Review comment:
Thank you for catching this. I double checked the FlinkKafkaConsumer and realized that it is disables the auto partition discovery by default. I have updated the PR to fix this.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 10afa89b37842487ecbcb3e4835d7919b35406c0 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lindong28 commented on a change in pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #15161:
URL: https://github.com/apache/flink/pull/15161#discussion_r600297493
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java
##########
@@ -88,11 +90,15 @@ public void init() throws Exception {
input = new StreamTaskSourceInput<>(sourceOperator, 0, 0);
}
+ CountingOutput<T> countingOutput =
Review comment:
Thank you Stephan for the explanation. Yes I agree we should try to minimize the number of wrappers on the performance critical path. I have updated the PR to move `counter` logic into `AsyncDataOutputToOutput`.
I noticed that [StreamTaskSourceOutput](https://github.com/apache/flink/blob/master/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamMultipleInputProcessorFactory.java#L225) currently does not take any metric whereas [StreamTaskNetworkOutput](https://github.com/apache/flink/blob/master/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamMultipleInputProcessorFactory.java#L205), whose usage is in the same loop body as `StreamTaskSourceOutput`, takes `mainOperatorRecordsIn` as input.
I tried to find the answer in the code. Still it is not clear to me whether `StreamTaskSourceOutput` should take metrics (e.g. numRecordsIn, numRecordsOut) as input and update them. I will keep the `StreamTaskSourceOutput` as is.
+@becketqin for comments.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "52088a88b0c5666ef2522248057de9f5a787452c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14779",
"triggerID" : "52088a88b0c5666ef2522248057de9f5a787452c",
"triggerType" : "PUSH"
}, {
"hash" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"triggerType" : "PUSH"
}, {
"hash" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15120",
"triggerID" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"triggerType" : "PUSH"
}, {
"hash" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15197",
"triggerID" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"triggerType" : "PUSH"
}, {
"hash" : "3bbb86b12a43044ce3f876401c98f28c87858fbe",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15366",
"triggerID" : "3bbb86b12a43044ce3f876401c98f28c87858fbe",
"triggerType" : "PUSH"
}, {
"hash" : "c58bb5faec3f07e0b66d0e042e3e9c6082fb8e9c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15416",
"triggerID" : "c58bb5faec3f07e0b66d0e042e3e9c6082fb8e9c",
"triggerType" : "PUSH"
}, {
"hash" : "6fc6cc9a2e6c26e9e8fa62b481237231394078eb",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15507",
"triggerID" : "6fc6cc9a2e6c26e9e8fa62b481237231394078eb",
"triggerType" : "PUSH"
}, {
"hash" : "6c17ec7241323224d3053134e82353c63f28a24f",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "6c17ec7241323224d3053134e82353c63f28a24f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d3f59e302b9860d49ce79252aeb8bd1decaeeabf UNKNOWN
* 6fc6cc9a2e6c26e9e8fa62b481237231394078eb Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15507)
* 6c17ec7241323224d3053134e82353c63f28a24f 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "52088a88b0c5666ef2522248057de9f5a787452c",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14779",
"triggerID" : "52088a88b0c5666ef2522248057de9f5a787452c",
"triggerType" : "PUSH"
}, {
"hash" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 52088a88b0c5666ef2522248057de9f5a787452c Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14779)
* d3f59e302b9860d49ce79252aeb8bd1decaeeabf 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lindong28 commented on a change in pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #15161:
URL: https://github.com/apache/flink/pull/15161#discussion_r600228593
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java
##########
@@ -442,10 +442,8 @@ private void parseAndSetRequiredProperties() {
true);
// If the source is bounded, do not run periodic partition discovery.
- if (maybeOverride(
- KafkaSourceOptions.PARTITION_DISCOVERY_INTERVAL_MS.key(),
- "-1",
- boundedness == Boundedness.BOUNDED)) {
+ if (boundedness == Boundedness.BOUNDED &&
Review comment:
Thank you for catching this. I double checked the FlinkKafkaConsumer and realized that it is disables the auto partition discovery by default. I have updated the PR to replace this commit with a new commit which "remove duplicated warning and remove redundant default value for partition.discovery.interval.ms".
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lindong28 commented on a change in pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #15161:
URL: https://github.com/apache/flink/pull/15161#discussion_r600297493
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java
##########
@@ -88,11 +90,15 @@ public void init() throws Exception {
input = new StreamTaskSourceInput<>(sourceOperator, 0, 0);
}
+ CountingOutput<T> countingOutput =
Review comment:
Thank you Stephan for the explanation. Yes I agree we should try to minimize the number of wrappers on the performance critical path. I have updated the PR to move `counter` logic into `AsyncDataOutputToOutput`.
As a separate but possibly related issue, I noticed that [StreamTaskSourceOutput](https://github.com/apache/flink/blob/master/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamMultipleInputProcessorFactory.java#L225) currently does not take any metric whereas [StreamTaskNetworkOutput](https://github.com/apache/flink/blob/master/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamMultipleInputProcessorFactory.java#L205), whose usage is in the same loop body as `StreamTaskSourceOutput`, takes `mainOperatorRecordsIn` as input.
I tried to find the answer in the code. Still it is not clear to me whether `StreamTaskSourceOutput` should take metrics (e.g. numRecordsIn, numRecordsOut) as input and update them. I will keep the `StreamTaskSourceOutput` as is.
+@becketqin for comments.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lindong28 removed a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
lindong28 removed a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-799917097
@flinkbot run azure
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lindong28 edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
lindong28 edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-806310235
@StephanEwen @becketqin Thanks for the review! I think all issues are resolved. I have removed from this PR the change related to the ClassNotFoundExceptionException.
Could you take another look?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14754",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "799917097",
"triggerType" : "MANUAL"
}, {
"hash" : "52088a88b0c5666ef2522248057de9f5a787452c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14779",
"triggerID" : "52088a88b0c5666ef2522248057de9f5a787452c",
"triggerType" : "PUSH"
}, {
"hash" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d3f59e302b9860d49ce79252aeb8bd1decaeeabf",
"triggerType" : "PUSH"
}, {
"hash" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15120",
"triggerID" : "728e9c6d1562fc250f942eaf147a3aacae95c436",
"triggerType" : "PUSH"
}, {
"hash" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15197",
"triggerID" : "871e9d892b6da0907cc3c5a1d811f145b0bb0f47",
"triggerType" : "PUSH"
}, {
"hash" : "3bbb86b12a43044ce3f876401c98f28c87858fbe",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3bbb86b12a43044ce3f876401c98f28c87858fbe",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d3f59e302b9860d49ce79252aeb8bd1decaeeabf UNKNOWN
* 871e9d892b6da0907cc3c5a1d811f145b0bb0f47 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=15197)
* 3bbb86b12a43044ce3f876401c98f28c87858fbe 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lindong28 commented on a change in pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #15161:
URL: https://github.com/apache/flink/pull/15161#discussion_r601981728
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java
##########
@@ -145,19 +155,25 @@ private void triggerCheckpointForExternallyInducedSource(long checkpointId) {
private final Output<StreamRecord<T>> output;
@Nullable private final WatermarkGauge inputWatermarkGauge;
+ @Nullable private final Counter numRecordsOut;
public AsyncDataOutputToOutput(
Output<StreamRecord<T>> output,
StreamStatusMaintainer streamStatusMaintainer,
- @Nullable WatermarkGauge inputWatermarkGauge) {
+ @Nullable WatermarkGauge inputWatermarkGauge,
+ @Nullable Counter numRecordsOut) {
super(streamStatusMaintainer);
this.output = checkNotNull(output);
this.inputWatermarkGauge = inputWatermarkGauge;
+ this.numRecordsOut = numRecordsOut;
}
@Override
public void emitRecord(StreamRecord<T> streamRecord) {
+ if (numRecordsOut != null) {
Review comment:
Sounds good. It is updated as suggested :)
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c45b554fee8aba36169cffe7b2da6621e7beb9d3 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661)
* 6b317a312546abfb536c1358a331f1878e65e5d3 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lindong28 commented on a change in pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #15161:
URL: https://github.com/apache/flink/pull/15161#discussion_r600194416
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContextTest.java
##########
@@ -182,6 +191,44 @@ public void testSnapshotAndRestore() throws Exception {
restoredTracker.assignmentsByCheckpointId());
}
+ @Test
+ public void testCallableInterruptedDuringShutdownDoNotFailJob() throws InterruptedException {
+ AtomicReference<Throwable> expectedError = new AtomicReference<>(null);
+
+ ManuallyTriggeredScheduledExecutorService manualWorkerExecutor =
+ new ManuallyTriggeredScheduledExecutorService();
+ ManuallyTriggeredScheduledExecutorService manualCoordinatorExecutor =
+ new ManuallyTriggeredScheduledExecutorService();
+
+ SourceCoordinatorContext<MockSourceSplit> testingContext =
+ new SourceCoordinatorContext<>(
+ manualCoordinatorExecutor,
+ manualWorkerExecutor,
+ new SourceCoordinatorProvider.CoordinatorExecutorThreadFactory(
+ TEST_OPERATOR_ID.toHexString(), getClass().getClassLoader()),
+ operatorCoordinatorContext,
+ new MockSourceSplitSerializer(),
+ splitSplitAssignmentTracker);
+
+ testingContext.callAsync(
+ () -> {
+ throw new InterruptedException();
+ },
+ (ignored, e) -> {
+ if (e != null) {
+ expectedError.set(e);
+ throw new RuntimeException(e);
+ }
+ });
+
+ manualWorkerExecutor.triggerAll();
Review comment:
It is needed here. `callAsync(...)` indirectly calls `manualWorkerExecutor.execute(...)`, which enqueues the runnable into an internal queue (in `ManuallyTriggeredScheduledExecutorService`) without actually executing the runnable.
We need to call `manualWorkerExecutor.triggerAll()` to execute the runnable.
I verified that, if we remove `manualWorkerExecutor.triggerAll()`, the test would fail due to `assertTrue(expectedError.get() instanceof InterruptedException)` failure.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
}, {
"hash" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14661",
"triggerID" : "c45b554fee8aba36169cffe7b2da6621e7beb9d3",
"triggerType" : "PUSH"
}, {
"hash" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692",
"triggerID" : "6b317a312546abfb536c1358a331f1878e65e5d3",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6b317a312546abfb536c1358a331f1878e65e5d3 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14692)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-797177953
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14508",
"triggerID" : "10afa89b37842487ecbcb3e4835d7919b35406c0",
"triggerType" : "PUSH"
}, {
"hash" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617",
"triggerID" : "2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2ecf0b62e16a219f12a07f8a8867cbfde0dbef2f Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14617)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] StephanEwen commented on pull request #15161: [FLINK-20114][connector/kafka,common] Fix a few KafkaSource-related bugs
Posted by GitBox <gi...@apache.org>.
StephanEwen commented on pull request #15161:
URL: https://github.com/apache/flink/pull/15161#issuecomment-804270036
Thanks for looking into these issues. I'll try to take a look asap so we can get these improvements into 1.13.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org