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/07/12 09:05:11 UTC
[GitHub] [flink] HuangXingBo opened a new pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
HuangXingBo opened a new pull request #16467:
URL: https://github.com/apache/flink/pull/16467
## What is the purpose of the change
*This pull request will optimize the finish bundle logic in PyFlink so that we can solve following three problems*
- *OutputStream can not receive any data in checkpointing*
- *Small data is hard to processed in pipeline*
- *Python UDTF output unbounded data*
## Brief change log
- *Move the blocking logic of finish bundle to async thread in Java Operator*
- *Use Schedule Task to flush data to in Python Side*
## Verifying this change
- *Manual Test to cover*
## 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, 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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16467:
URL: https://github.com/apache/flink/pull/16467#issuecomment-878120698
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20320",
"triggerID" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"triggerType" : "PUSH"
}, {
"hash" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20375",
"triggerID" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"triggerType" : "PUSH"
}, {
"hash" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20404",
"triggerID" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"triggerType" : "PUSH"
}, {
"hash" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20717",
"triggerID" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 32cd0aa95dba002136650297a83caf1edb5ae3b7 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20717)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16467:
URL: https://github.com/apache/flink/pull/16467#issuecomment-878120698
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20320",
"triggerID" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"triggerType" : "PUSH"
}, {
"hash" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20375",
"triggerID" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"triggerType" : "PUSH"
}, {
"hash" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20404",
"triggerID" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"triggerType" : "PUSH"
}, {
"hash" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20717",
"triggerID" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"triggerType" : "PUSH"
}, {
"hash" : "76064d25e35a22383d86db691bec3a2f3c89374f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21099",
"triggerID" : "76064d25e35a22383d86db691bec3a2f3c89374f",
"triggerType" : "PUSH"
}, {
"hash" : "4ad0ed97e37a2d1926bedf1fe1e09e9b5b3cdd9a",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21944",
"triggerID" : "4ad0ed97e37a2d1926bedf1fe1e09e9b5b3cdd9a",
"triggerType" : "PUSH"
}, {
"hash" : "0faa73985a670f833c4475559600ea01cdb579f1",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21958",
"triggerID" : "0faa73985a670f833c4475559600ea01cdb579f1",
"triggerType" : "PUSH"
}, {
"hash" : "a74e0e246e97b0e23e01017f006079645bc14093",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21973",
"triggerID" : "a74e0e246e97b0e23e01017f006079645bc14093",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4ad0ed97e37a2d1926bedf1fe1e09e9b5b3cdd9a Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21944)
* 0faa73985a670f833c4475559600ea01cdb579f1 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21958)
* a74e0e246e97b0e23e01017f006079645bc14093 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21973)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16467:
URL: https://github.com/apache/flink/pull/16467#issuecomment-878120698
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20320",
"triggerID" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"triggerType" : "PUSH"
}, {
"hash" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20375",
"triggerID" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"triggerType" : "PUSH"
}, {
"hash" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20404",
"triggerID" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"triggerType" : "PUSH"
}, {
"hash" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* fd94525e8bccd478a175e531074a1356257e0dcf Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20404)
* 32cd0aa95dba002136650297a83caf1edb5ae3b7 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dianfu closed pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
dianfu closed pull request #16467:
URL: https://github.com/apache/flink/pull/16467
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16467:
URL: https://github.com/apache/flink/pull/16467#issuecomment-878120698
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20320",
"triggerID" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"triggerType" : "PUSH"
}, {
"hash" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20375",
"triggerID" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"triggerType" : "PUSH"
}, {
"hash" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20404",
"triggerID" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7a4f6093c1b70695a471de482631ce28f527eb88 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20375)
* fd94525e8bccd478a175e531074a1356257e0dcf Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20404)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16467:
URL: https://github.com/apache/flink/pull/16467#issuecomment-878120698
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16467:
URL: https://github.com/apache/flink/pull/16467#issuecomment-878120698
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20320",
"triggerID" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"triggerType" : "PUSH"
}, {
"hash" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20375",
"triggerID" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"triggerType" : "PUSH"
}, {
"hash" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20404",
"triggerID" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* fd94525e8bccd478a175e531074a1356257e0dcf Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20404)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16467:
URL: https://github.com/apache/flink/pull/16467#issuecomment-878120698
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20320",
"triggerID" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"triggerType" : "PUSH"
}, {
"hash" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b4bd8ccfe991db22837aa2f7e589a052bc6320cf Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20320)
* 7a4f6093c1b70695a471de482631ce28f527eb88 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16467:
URL: https://github.com/apache/flink/pull/16467#issuecomment-878120698
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16467:
URL: https://github.com/apache/flink/pull/16467#issuecomment-878120698
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20320",
"triggerID" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"triggerType" : "PUSH"
}, {
"hash" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20375",
"triggerID" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"triggerType" : "PUSH"
}, {
"hash" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20404",
"triggerID" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"triggerType" : "PUSH"
}, {
"hash" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20717",
"triggerID" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"triggerType" : "PUSH"
}, {
"hash" : "76064d25e35a22383d86db691bec3a2f3c89374f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21099",
"triggerID" : "76064d25e35a22383d86db691bec3a2f3c89374f",
"triggerType" : "PUSH"
}, {
"hash" : "4ad0ed97e37a2d1926bedf1fe1e09e9b5b3cdd9a",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21944",
"triggerID" : "4ad0ed97e37a2d1926bedf1fe1e09e9b5b3cdd9a",
"triggerType" : "PUSH"
}, {
"hash" : "0faa73985a670f833c4475559600ea01cdb579f1",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21958",
"triggerID" : "0faa73985a670f833c4475559600ea01cdb579f1",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4ad0ed97e37a2d1926bedf1fe1e09e9b5b3cdd9a Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21944)
* 0faa73985a670f833c4475559600ea01cdb579f1 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21958)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16467:
URL: https://github.com/apache/flink/pull/16467#issuecomment-878120698
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20320",
"triggerID" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"triggerType" : "PUSH"
}, {
"hash" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20375",
"triggerID" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"triggerType" : "PUSH"
}, {
"hash" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20404",
"triggerID" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"triggerType" : "PUSH"
}, {
"hash" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20717",
"triggerID" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"triggerType" : "PUSH"
}, {
"hash" : "76064d25e35a22383d86db691bec3a2f3c89374f",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "76064d25e35a22383d86db691bec3a2f3c89374f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 32cd0aa95dba002136650297a83caf1edb5ae3b7 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20717)
* 76064d25e35a22383d86db691bec3a2f3c89374f UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16467:
URL: https://github.com/apache/flink/pull/16467#issuecomment-878120698
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20320",
"triggerID" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"triggerType" : "PUSH"
}, {
"hash" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20375",
"triggerID" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"triggerType" : "PUSH"
}, {
"hash" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20404",
"triggerID" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"triggerType" : "PUSH"
}, {
"hash" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20717",
"triggerID" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"triggerType" : "PUSH"
}, {
"hash" : "76064d25e35a22383d86db691bec3a2f3c89374f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21099",
"triggerID" : "76064d25e35a22383d86db691bec3a2f3c89374f",
"triggerType" : "PUSH"
}, {
"hash" : "4ad0ed97e37a2d1926bedf1fe1e09e9b5b3cdd9a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21944",
"triggerID" : "4ad0ed97e37a2d1926bedf1fe1e09e9b5b3cdd9a",
"triggerType" : "PUSH"
}, {
"hash" : "0faa73985a670f833c4475559600ea01cdb579f1",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21958",
"triggerID" : "0faa73985a670f833c4475559600ea01cdb579f1",
"triggerType" : "PUSH"
}, {
"hash" : "a74e0e246e97b0e23e01017f006079645bc14093",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21973",
"triggerID" : "a74e0e246e97b0e23e01017f006079645bc14093",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0faa73985a670f833c4475559600ea01cdb579f1 Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21958)
* a74e0e246e97b0e23e01017f006079645bc14093 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21973)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16467:
URL: https://github.com/apache/flink/pull/16467#issuecomment-878120698
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20320",
"triggerID" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"triggerType" : "PUSH"
}, {
"hash" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20375",
"triggerID" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7a4f6093c1b70695a471de482631ce28f527eb88 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20375)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16467:
URL: https://github.com/apache/flink/pull/16467#issuecomment-878120698
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20320",
"triggerID" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"triggerType" : "PUSH"
}, {
"hash" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20375",
"triggerID" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"triggerType" : "PUSH"
}, {
"hash" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7a4f6093c1b70695a471de482631ce28f527eb88 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20375)
* fd94525e8bccd478a175e531074a1356257e0dcf UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16467:
URL: https://github.com/apache/flink/pull/16467#issuecomment-878120698
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16467:
URL: https://github.com/apache/flink/pull/16467#issuecomment-878120698
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20320",
"triggerID" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"triggerType" : "PUSH"
}, {
"hash" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20375",
"triggerID" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"triggerType" : "PUSH"
}, {
"hash" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20404",
"triggerID" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"triggerType" : "PUSH"
}, {
"hash" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20717",
"triggerID" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"triggerType" : "PUSH"
}, {
"hash" : "76064d25e35a22383d86db691bec3a2f3c89374f",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21099",
"triggerID" : "76064d25e35a22383d86db691bec3a2f3c89374f",
"triggerType" : "PUSH"
}, {
"hash" : "4ad0ed97e37a2d1926bedf1fe1e09e9b5b3cdd9a",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21944",
"triggerID" : "4ad0ed97e37a2d1926bedf1fe1e09e9b5b3cdd9a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 76064d25e35a22383d86db691bec3a2f3c89374f Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21099)
* 4ad0ed97e37a2d1926bedf1fe1e09e9b5b3cdd9a Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21944)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16467:
URL: https://github.com/apache/flink/pull/16467#issuecomment-878120698
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20320",
"triggerID" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"triggerType" : "PUSH"
}, {
"hash" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20375",
"triggerID" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"triggerType" : "PUSH"
}, {
"hash" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20404",
"triggerID" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"triggerType" : "PUSH"
}, {
"hash" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20717",
"triggerID" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"triggerType" : "PUSH"
}, {
"hash" : "76064d25e35a22383d86db691bec3a2f3c89374f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21099",
"triggerID" : "76064d25e35a22383d86db691bec3a2f3c89374f",
"triggerType" : "PUSH"
}, {
"hash" : "4ad0ed97e37a2d1926bedf1fe1e09e9b5b3cdd9a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21944",
"triggerID" : "4ad0ed97e37a2d1926bedf1fe1e09e9b5b3cdd9a",
"triggerType" : "PUSH"
}, {
"hash" : "0faa73985a670f833c4475559600ea01cdb579f1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21958",
"triggerID" : "0faa73985a670f833c4475559600ea01cdb579f1",
"triggerType" : "PUSH"
}, {
"hash" : "a74e0e246e97b0e23e01017f006079645bc14093",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21973",
"triggerID" : "a74e0e246e97b0e23e01017f006079645bc14093",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a74e0e246e97b0e23e01017f006079645bc14093 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21973)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dianfu commented on a change in pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
dianfu commented on a change in pull request #16467:
URL: https://github.com/apache/flink/pull/16467#discussion_r687486737
##########
File path: flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java
##########
@@ -317,9 +329,41 @@ private void checkInvokeFinishBundleByTime() throws Exception {
protected void invokeFinishBundle() throws Exception {
if (elementCount > 0) {
- pythonFunctionRunner.flush();
- elementCount = 0;
+ AtomicBoolean flushThreadFinish = new AtomicBoolean(false);
+ AtomicReference<Throwable> exceptionReference = new AtomicReference<>();
+ flushThreadPool.submit(
+ () -> {
+ try {
+ pythonFunctionRunner.flush();
+ } catch (Throwable e) {
+ exceptionReference.set(e);
+ } finally {
+ flushThreadFinish.set(true);
+ // interrupt the progress of takeResult to avoid the main thread is
+ // blocked forever.
+ ((BeamPythonFunctionRunner) pythonFunctionRunner).notifyNoMoreResults();
+ }
+ });
+ Tuple2<byte[], Integer> resultTuple;
+ while (!flushThreadFinish.get()) {
+ resultTuple = pythonFunctionRunner.takeResult();
+ if (resultTuple.f1 != 0) {
+ emitResult(resultTuple);
+ emitResults();
+ }
+ }
emitResults();
+ Throwable flushThreadThrowable = exceptionReference.get();
+ if (flushThreadThrowable != null) {
Review comment:
if (flushThreadThrowable != null) {
throw new Exception(flushThreadThrowable);
}
##########
File path: flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java
##########
@@ -158,6 +167,9 @@ public void close() throws Exception {
pythonFunctionRunner.close();
pythonFunctionRunner = null;
}
+ if (flushThreadPool != null) {
+ flushThreadPool.shutdown();
Review comment:
flushThreadPool = null
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16467:
URL: https://github.com/apache/flink/pull/16467#issuecomment-878120698
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20320",
"triggerID" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"triggerType" : "PUSH"
}, {
"hash" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20375",
"triggerID" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"triggerType" : "PUSH"
}, {
"hash" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20404",
"triggerID" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"triggerType" : "PUSH"
}, {
"hash" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20717",
"triggerID" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"triggerType" : "PUSH"
}, {
"hash" : "76064d25e35a22383d86db691bec3a2f3c89374f",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21099",
"triggerID" : "76064d25e35a22383d86db691bec3a2f3c89374f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 76064d25e35a22383d86db691bec3a2f3c89374f Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21099)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16467:
URL: https://github.com/apache/flink/pull/16467#issuecomment-878120698
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20320",
"triggerID" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"triggerType" : "PUSH"
}, {
"hash" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20375",
"triggerID" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"triggerType" : "PUSH"
}, {
"hash" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20404",
"triggerID" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"triggerType" : "PUSH"
}, {
"hash" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20717",
"triggerID" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"triggerType" : "PUSH"
}, {
"hash" : "76064d25e35a22383d86db691bec3a2f3c89374f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21099",
"triggerID" : "76064d25e35a22383d86db691bec3a2f3c89374f",
"triggerType" : "PUSH"
}, {
"hash" : "4ad0ed97e37a2d1926bedf1fe1e09e9b5b3cdd9a",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21944",
"triggerID" : "4ad0ed97e37a2d1926bedf1fe1e09e9b5b3cdd9a",
"triggerType" : "PUSH"
}, {
"hash" : "0faa73985a670f833c4475559600ea01cdb579f1",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21958",
"triggerID" : "0faa73985a670f833c4475559600ea01cdb579f1",
"triggerType" : "PUSH"
}, {
"hash" : "a74e0e246e97b0e23e01017f006079645bc14093",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "a74e0e246e97b0e23e01017f006079645bc14093",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4ad0ed97e37a2d1926bedf1fe1e09e9b5b3cdd9a Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21944)
* 0faa73985a670f833c4475559600ea01cdb579f1 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21958)
* a74e0e246e97b0e23e01017f006079645bc14093 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16467:
URL: https://github.com/apache/flink/pull/16467#issuecomment-878120698
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20320",
"triggerID" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"triggerType" : "PUSH"
}, {
"hash" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20375",
"triggerID" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"triggerType" : "PUSH"
}, {
"hash" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20404",
"triggerID" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"triggerType" : "PUSH"
}, {
"hash" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20717",
"triggerID" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* fd94525e8bccd478a175e531074a1356257e0dcf Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20404)
* 32cd0aa95dba002136650297a83caf1edb5ae3b7 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20717)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16467:
URL: https://github.com/apache/flink/pull/16467#issuecomment-878120698
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20320",
"triggerID" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"triggerType" : "PUSH"
}, {
"hash" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20375",
"triggerID" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"triggerType" : "PUSH"
}, {
"hash" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20404",
"triggerID" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"triggerType" : "PUSH"
}, {
"hash" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20717",
"triggerID" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"triggerType" : "PUSH"
}, {
"hash" : "76064d25e35a22383d86db691bec3a2f3c89374f",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21099",
"triggerID" : "76064d25e35a22383d86db691bec3a2f3c89374f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 32cd0aa95dba002136650297a83caf1edb5ae3b7 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20717)
* 76064d25e35a22383d86db691bec3a2f3c89374f Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21099)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16467:
URL: https://github.com/apache/flink/pull/16467#issuecomment-878120698
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20320",
"triggerID" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"triggerType" : "PUSH"
}, {
"hash" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20375",
"triggerID" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"triggerType" : "PUSH"
}, {
"hash" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20404",
"triggerID" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"triggerType" : "PUSH"
}, {
"hash" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20717",
"triggerID" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"triggerType" : "PUSH"
}, {
"hash" : "76064d25e35a22383d86db691bec3a2f3c89374f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21099",
"triggerID" : "76064d25e35a22383d86db691bec3a2f3c89374f",
"triggerType" : "PUSH"
}, {
"hash" : "4ad0ed97e37a2d1926bedf1fe1e09e9b5b3cdd9a",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21944",
"triggerID" : "4ad0ed97e37a2d1926bedf1fe1e09e9b5b3cdd9a",
"triggerType" : "PUSH"
}, {
"hash" : "0faa73985a670f833c4475559600ea01cdb579f1",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "0faa73985a670f833c4475559600ea01cdb579f1",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4ad0ed97e37a2d1926bedf1fe1e09e9b5b3cdd9a Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21944)
* 0faa73985a670f833c4475559600ea01cdb579f1 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16467:
URL: https://github.com/apache/flink/pull/16467#issuecomment-878120698
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20320",
"triggerID" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"triggerType" : "PUSH"
}, {
"hash" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20375",
"triggerID" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b4bd8ccfe991db22837aa2f7e589a052bc6320cf Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20320)
* 7a4f6093c1b70695a471de482631ce28f527eb88 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20375)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] HuangXingBo commented on a change in pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
HuangXingBo commented on a change in pull request #16467:
URL: https://github.com/apache/flink/pull/16467#discussion_r687329191
##########
File path: flink-python/pyflink/fn_execution/stream_fast.pyx
##########
@@ -94,6 +100,21 @@ cdef class OutputStream:
self.buffer_size = 1024
self.pos = 0
+ cpdef void write(self, bytes v):
Review comment:
This method could be called in pure python function e.g. `ResettableIO` which is used in `ArrowCoderImpl`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16467:
URL: https://github.com/apache/flink/pull/16467#issuecomment-878120698
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20320",
"triggerID" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"triggerType" : "PUSH"
}, {
"hash" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20375",
"triggerID" : "7a4f6093c1b70695a471de482631ce28f527eb88",
"triggerType" : "PUSH"
}, {
"hash" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20404",
"triggerID" : "fd94525e8bccd478a175e531074a1356257e0dcf",
"triggerType" : "PUSH"
}, {
"hash" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=20717",
"triggerID" : "32cd0aa95dba002136650297a83caf1edb5ae3b7",
"triggerType" : "PUSH"
}, {
"hash" : "76064d25e35a22383d86db691bec3a2f3c89374f",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21099",
"triggerID" : "76064d25e35a22383d86db691bec3a2f3c89374f",
"triggerType" : "PUSH"
}, {
"hash" : "4ad0ed97e37a2d1926bedf1fe1e09e9b5b3cdd9a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4ad0ed97e37a2d1926bedf1fe1e09e9b5b3cdd9a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 76064d25e35a22383d86db691bec3a2f3c89374f Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21099)
* 4ad0ed97e37a2d1926bedf1fe1e09e9b5b3cdd9a UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #16467:
URL: https://github.com/apache/flink/pull/16467#issuecomment-878120698
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "b4bd8ccfe991db22837aa2f7e589a052bc6320cf",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b4bd8ccfe991db22837aa2f7e589a052bc6320cf UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dianfu commented on a change in pull request #16467: [FLINK-23309][python] Optimize the finish bundle logic in PyFlink
Posted by GitBox <gi...@apache.org>.
dianfu commented on a change in pull request #16467:
URL: https://github.com/apache/flink/pull/16467#discussion_r686617600
##########
File path: flink-python/pyflink/fn_execution/beam/beam_coder_impl_fast.pyx
##########
@@ -94,10 +94,13 @@ cdef class FlinkLengthPrefixCoderBeamWrapper(StreamCoderImpl):
"""
def __cinit__(self, value_coder):
self._value_coder = value_coder
+ self._output_stream = None
cpdef encode_to_stream(self, value, BOutputStream out_stream, bint nested):
- output_stream = BeamOutputStream(out_stream)
- self._value_coder.encode_to_stream(value, output_stream)
+ if not self._output_stream:
+ self._output_stream = BeamTimeBasedOutputStream()
Review comment:
Why not initializing **_output_stream** in the constructor?
##########
File path: flink-python/pyflink/fn_execution/beam/beam_coder_impl_slow.py
##########
@@ -58,4 +58,25 @@ def decode_from_stream(self, in_stream: create_InputStream, nested):
return self._value_coder.decode_from_stream(data_input_stream)
def __repr__(self):
- return 'FlinkCoderBeamWrapper[%s]' % self._value_coder
+ return 'FlinkFieldCoderBeamWrapper[%s]' % self._value_coder
+
+
+class FlinkLengthPrefixCoderBeamWrapper(FlinkFieldCoderBeamWrapper):
+ """
+ Bridge between Beam coder and Flink coder for the top-level LengthPrefixCoder.
+ """
+ def __init__(self, value_coder):
+ super(FlinkLengthPrefixCoderBeamWrapper, self).__init__(value_coder)
+ self._output_stream = None
Review comment:
Could we give it a more meaningful name? It's difficult to differentiate between _output_stream and out_stream.
##########
File path: flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java
##########
@@ -317,9 +321,41 @@ private void checkInvokeFinishBundleByTime() throws Exception {
protected void invokeFinishBundle() throws Exception {
if (elementCount > 0) {
- pythonFunctionRunner.flush();
- elementCount = 0;
+ AtomicBoolean flushThreadFinish = new AtomicBoolean(false);
+ CountDownLatch flushThreadStart = new CountDownLatch(1);
+ AtomicReference<Exception> exceptionReference = new AtomicReference<>();
+ Thread flushThread =
Review comment:
It's would be better to create a single thread pool for this, e.g. Executors.newSingleThreadExecutor. It could avoid creating a new thread for each bundle.
##########
File path: flink-python/pyflink/fn_execution/beam/beam_stream_slow.py
##########
@@ -33,3 +34,31 @@ def read_byte(self):
def size(self):
return self._input_stream.size()
+
+
+class BeamTimeBasedOutputStream(create_OutputStream):
+ def __init__(self):
+ super(BeamTimeBasedOutputStream).__init__()
+ self._flush_event = False
+ self._periodic_flusher = PeriodicThread(1, self.notify_flush)
+ self._periodic_flusher.daemon = True
+ self._periodic_flusher.start()
+ self._output_stream = None
+
+ def write(self, b: bytes):
+ self._output_stream.write(b)
+ if self._flush_event:
+ self._output_stream.flush()
+ self._flush_event = False
+
+ def parse_output_stream(self, output_stream: create_OutputStream):
+ self._output_stream = output_stream
+
+ def notify_flush(self):
+ if not self._flush_event:
Review comment:
```suggestion
if not self._flush_event:
```
I guess this line could be removed
##########
File path: flink-python/pyflink/fn_execution/beam/beam_stream_slow.py
##########
@@ -33,3 +34,31 @@ def read_byte(self):
def size(self):
return self._input_stream.size()
+
+
+class BeamTimeBasedOutputStream(create_OutputStream):
+ def __init__(self):
+ super(BeamTimeBasedOutputStream).__init__()
+ self._flush_event = False
+ self._periodic_flusher = PeriodicThread(1, self.notify_flush)
+ self._periodic_flusher.daemon = True
+ self._periodic_flusher.start()
+ self._output_stream = None
+
+ def write(self, b: bytes):
+ self._output_stream.write(b)
+ if self._flush_event:
+ self._output_stream.flush()
+ self._flush_event = False
+
+ def parse_output_stream(self, output_stream: create_OutputStream):
Review comment:
```suggestion
def reset_output_stream(self, output_stream: create_OutputStream):
```
##########
File path: flink-python/pyflink/fn_execution/stream_fast.pyx
##########
@@ -94,6 +100,21 @@ cdef class OutputStream:
self.buffer_size = 1024
self.pos = 0
+ cpdef void write(self, bytes v):
Review comment:
Why it's cpdef and the other methods are declared cdef?
##########
File path: flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java
##########
@@ -317,9 +321,41 @@ private void checkInvokeFinishBundleByTime() throws Exception {
protected void invokeFinishBundle() throws Exception {
if (elementCount > 0) {
- pythonFunctionRunner.flush();
- elementCount = 0;
+ AtomicBoolean flushThreadFinish = new AtomicBoolean(false);
+ CountDownLatch flushThreadStart = new CountDownLatch(1);
+ AtomicReference<Exception> exceptionReference = new AtomicReference<>();
+ Thread flushThread =
+ new Thread(
+ () -> {
+ try {
+ flushThreadStart.countDown();
+ pythonFunctionRunner.flush();
+ } catch (Exception e) {
+ exceptionReference.set(e);
+ } finally {
+ flushThreadFinish.set(true);
+ // interrupt the progress of takeResult in avoid of the main
+ // thread is locked forever.
Review comment:
```suggestion
// thread is blocked forever.
```
##########
File path: flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java
##########
@@ -317,9 +321,41 @@ private void checkInvokeFinishBundleByTime() throws Exception {
protected void invokeFinishBundle() throws Exception {
if (elementCount > 0) {
- pythonFunctionRunner.flush();
- elementCount = 0;
+ AtomicBoolean flushThreadFinish = new AtomicBoolean(false);
+ CountDownLatch flushThreadStart = new CountDownLatch(1);
+ AtomicReference<Exception> exceptionReference = new AtomicReference<>();
+ Thread flushThread =
+ new Thread(
+ () -> {
+ try {
+ flushThreadStart.countDown();
+ pythonFunctionRunner.flush();
+ } catch (Exception e) {
+ exceptionReference.set(e);
+ } finally {
+ flushThreadFinish.set(true);
+ // interrupt the progress of takeResult in avoid of the main
+ // thread is locked forever.
+ ((BeamPythonFunctionRunner) pythonFunctionRunner)
+ .noEmptySignal();
Review comment:
```suggestion
.notifyNoMoreResults();
```
##########
File path: flink-python/pyflink/fn_execution/beam/beam_operations_slow.py
##########
@@ -93,6 +94,7 @@ def process(self, o: WindowedValue):
self._value_coder_impl.encode_to_stream(
self.process_element(value), output_stream, True)
output_stream.maybe_flush()
+ self._value_coder_impl._output_stream.close()
Review comment:
When it's opened/started?
##########
File path: flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java
##########
@@ -317,9 +321,41 @@ private void checkInvokeFinishBundleByTime() throws Exception {
protected void invokeFinishBundle() throws Exception {
if (elementCount > 0) {
- pythonFunctionRunner.flush();
- elementCount = 0;
+ AtomicBoolean flushThreadFinish = new AtomicBoolean(false);
+ CountDownLatch flushThreadStart = new CountDownLatch(1);
+ AtomicReference<Exception> exceptionReference = new AtomicReference<>();
+ Thread flushThread =
+ new Thread(
+ () -> {
+ try {
+ flushThreadStart.countDown();
+ pythonFunctionRunner.flush();
+ } catch (Exception e) {
+ exceptionReference.set(e);
+ } finally {
+ flushThreadFinish.set(true);
+ // interrupt the progress of takeResult in avoid of the main
+ // thread is locked forever.
+ ((BeamPythonFunctionRunner) pythonFunctionRunner)
+ .noEmptySignal();
+ }
+ });
+ flushThread.start();
+ flushThreadStart.await();
Review comment:
Why need to wait the flush thread started?
##########
File path: flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java
##########
@@ -317,9 +321,41 @@ private void checkInvokeFinishBundleByTime() throws Exception {
protected void invokeFinishBundle() throws Exception {
if (elementCount > 0) {
- pythonFunctionRunner.flush();
- elementCount = 0;
+ AtomicBoolean flushThreadFinish = new AtomicBoolean(false);
+ CountDownLatch flushThreadStart = new CountDownLatch(1);
+ AtomicReference<Exception> exceptionReference = new AtomicReference<>();
+ Thread flushThread =
+ new Thread(
+ () -> {
+ try {
+ flushThreadStart.countDown();
+ pythonFunctionRunner.flush();
+ } catch (Exception e) {
+ exceptionReference.set(e);
+ } finally {
+ flushThreadFinish.set(true);
+ // interrupt the progress of takeResult in avoid of the main
Review comment:
```suggestion
// interrupt the progress of takeResult to avoid the main
```
##########
File path: flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java
##########
@@ -317,9 +321,41 @@ private void checkInvokeFinishBundleByTime() throws Exception {
protected void invokeFinishBundle() throws Exception {
if (elementCount > 0) {
- pythonFunctionRunner.flush();
- elementCount = 0;
+ AtomicBoolean flushThreadFinish = new AtomicBoolean(false);
+ CountDownLatch flushThreadStart = new CountDownLatch(1);
+ AtomicReference<Exception> exceptionReference = new AtomicReference<>();
+ Thread flushThread =
+ new Thread(
+ () -> {
+ try {
+ flushThreadStart.countDown();
+ pythonFunctionRunner.flush();
+ } catch (Exception e) {
Review comment:
```suggestion
} catch (Throwable t) {
```
##########
File path: flink-python/pyflink/fn_execution/beam/beam_stream_fast.pyx
##########
@@ -91,12 +90,33 @@ cdef class BeamOutputStream(LengthPrefixOutputStream):
self._output_stream.flush()
self._output_pos = 0
- cdef void _parse_output_stream(self, BOutputStream output_stream):
+ cdef void parse_output_stream(self, BOutputStream output_stream):
+ self._output_stream = output_stream
self._output_data = output_stream.data
self._output_pos = output_stream.pos
self._output_buffer_size = output_stream.buffer_size
cdef void _maybe_flush(self):
if self._output_pos > 10_000_000:
- self._output_stream.flush()
- self._output_pos = 0
+ self.flush()
+
+cdef class BeamTimeBasedOutputStream(BeamSizeBasedOutputStream):
+ def __init__(self, *args, **kwargs):
+ self._flush_event = False
+ self._periodic_flusher = PeriodicThread(1, self.notify_flush)
+ self._periodic_flusher.daemon = True
+ self._periodic_flusher.start()
+
+ cpdef void notify_flush(self):
+ if not self._flush_event:
+ self._flush_event = True
+
+ cdef void _maybe_flush(self):
+ if self._flush_event or self._output_pos > 10_000_000:
Review comment:
```suggestion
if self._flush_event or self._output_pos > 10_000_000:
```
Why not calling the method _maybe_flush defined in the parent class?
--
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