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 2022/02/17 11:17:08 UTC
[GitHub] [flink] imaffe opened a new pull request #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
imaffe opened a new pull request #18816:
URL: https://github.com/apache/flink/pull/18816
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
Add pulsar sink metrics and tests. Documentation will be included in a different PR with Chinese version
## Brief change log
- Standard sink metrics are supported.
- currentSendTime is the time from sendAsync() called to ack time, it's not available in NONE delivery mode (as there is no complete callback)
- sendLatencies is added on a per producer basis
## Verifying this change
Please make sure both new and modified tests in this PR follows the conventions defined in our code quality guide: https://flink.apache.org/contributing/code-style-and-quality-common.html#testing
This change added tests and can be verified as follows:
- Added tests show that `lastSendTime` can be modified
- Added tests showing metrics are present after writer creation
- Added tests for Counter and Gauge (maxLatency)type tests
However, per producer gauges (mainly latencies) are not tested, however their implementation is very similar to the maxLatency implementation
## Does this pull request potentially affect one of the following parts:
- No
## Documentation
Documentation is needed, documentation will be added after FLINK-26028
--
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] imaffe commented on a change in pull request #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
imaffe commented on a change in pull request #18816:
URL: https://github.com/apache/flink/pull/18816#discussion_r809602712
##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java
##########
@@ -124,8 +131,15 @@ public PulsarWriter(
throw new FlinkRuntimeException("Cannot initialize schema.", e);
}
+ // Initialize metrics
+ this.pulsarSinkWriterMetrics = new PulsarSinkWriterMetrics(initContext.metricGroup());
+ this.lastSendTime = PulsarSinkWriterMetrics.INVALID_LAST_SEND_TIME;
// Create this producer register after opening serialization schema!
- this.producerRegister = new TopicProducerRegister(sinkConfiguration);
+ this.producerRegister =
+ new TopicProducerRegister(sinkConfiguration, pulsarSinkWriterMetrics);
Review comment:
Consider keep pulsasrSinkWriterMetrics outside of topicProducerRegister, instead, all metrics registering code should happen in pulsarWriter.
--
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 #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18816:
URL: https://github.com/apache/flink/pull/18816#issuecomment-1042851613
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31751",
"triggerID" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"triggerType" : "PUSH"
}, {
"hash" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31787",
"triggerID" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0229cbdda1c15c851c2b594b721e388b9b0b7e5f Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31751)
* 6b17fb81597d3d67dcd2367659245cc01707fec2 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31787)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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 #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18816:
URL: https://github.com/apache/flink/pull/18816#issuecomment-1042851613
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31751",
"triggerID" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0229cbdda1c15c851c2b594b721e388b9b0b7e5f Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31751)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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 #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18816:
URL: https://github.com/apache/flink/pull/18816#issuecomment-1042851613
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31751",
"triggerID" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"triggerType" : "PUSH"
}, {
"hash" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31787",
"triggerID" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"triggerType" : "PUSH"
}, {
"hash" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31924",
"triggerID" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"triggerType" : "PUSH"
}, {
"hash" : "271648472e6f8e46df9945f096666a65d34f762c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31949",
"triggerID" : "271648472e6f8e46df9945f096666a65d34f762c",
"triggerType" : "PUSH"
}, {
"hash" : "26366818f1c6a7b29ddb063414d1b3a2eaf34408",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33065",
"triggerID" : "26366818f1c6a7b29ddb063414d1b3a2eaf34408",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 271648472e6f8e46df9945f096666a65d34f762c Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31949)
* 26366818f1c6a7b29ddb063414d1b3a2eaf34408 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33065)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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 #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18816:
URL: https://github.com/apache/flink/pull/18816#issuecomment-1042851613
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31751",
"triggerID" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"triggerType" : "PUSH"
}, {
"hash" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31787",
"triggerID" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"triggerType" : "PUSH"
}, {
"hash" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31924",
"triggerID" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"triggerType" : "PUSH"
}, {
"hash" : "271648472e6f8e46df9945f096666a65d34f762c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31949",
"triggerID" : "271648472e6f8e46df9945f096666a65d34f762c",
"triggerType" : "PUSH"
}, {
"hash" : "26366818f1c6a7b29ddb063414d1b3a2eaf34408",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "26366818f1c6a7b29ddb063414d1b3a2eaf34408",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 271648472e6f8e46df9945f096666a65d34f762c Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31949)
* 26366818f1c6a7b29ddb063414d1b3a2eaf34408 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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 #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18816:
URL: https://github.com/apache/flink/pull/18816#issuecomment-1042851613
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31751",
"triggerID" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0229cbdda1c15c851c2b594b721e388b9b0b7e5f Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31751)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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] imaffe commented on a change in pull request #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
imaffe commented on a change in pull request #18816:
URL: https://github.com/apache/flink/pull/18816#discussion_r811612902
##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java
##########
@@ -124,8 +131,17 @@ public PulsarWriter(
throw new FlinkRuntimeException("Cannot initialize schema.", e);
}
+ // Initialize metrics
+ this.pulsarSinkWriterMetricUtils =
Review comment:
The naming is a bit confusing, it should not be utils tho, need to come up with a better naming
--
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 #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18816:
URL: https://github.com/apache/flink/pull/18816#issuecomment-1042851613
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31751",
"triggerID" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"triggerType" : "PUSH"
}, {
"hash" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31787",
"triggerID" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"triggerType" : "PUSH"
}, {
"hash" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6b17fb81597d3d67dcd2367659245cc01707fec2 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31787)
* b7832f41b7aed0a43855fc2a79df01783802b4b4 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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 #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18816:
URL: https://github.com/apache/flink/pull/18816#issuecomment-1042851613
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31751",
"triggerID" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"triggerType" : "PUSH"
}, {
"hash" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0229cbdda1c15c851c2b594b721e388b9b0b7e5f Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31751)
* 6b17fb81597d3d67dcd2367659245cc01707fec2 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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] syhily commented on pull request #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
syhily commented on pull request #18816:
URL: https://github.com/apache/flink/pull/18816#issuecomment-1069082094
@imaffe Can you make this ci green?
--
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 #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18816:
URL: https://github.com/apache/flink/pull/18816#issuecomment-1042851613
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31751",
"triggerID" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"triggerType" : "PUSH"
}, {
"hash" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31787",
"triggerID" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"triggerType" : "PUSH"
}, {
"hash" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31924",
"triggerID" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"triggerType" : "PUSH"
}, {
"hash" : "271648472e6f8e46df9945f096666a65d34f762c",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31949",
"triggerID" : "271648472e6f8e46df9945f096666a65d34f762c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b7832f41b7aed0a43855fc2a79df01783802b4b4 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31924)
* 271648472e6f8e46df9945f096666a65d34f762c Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31949)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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 #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18816:
URL: https://github.com/apache/flink/pull/18816#issuecomment-1042851613
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31751",
"triggerID" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"triggerType" : "PUSH"
}, {
"hash" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31787",
"triggerID" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"triggerType" : "PUSH"
}, {
"hash" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31924",
"triggerID" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b7832f41b7aed0a43855fc2a79df01783802b4b4 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31924)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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] syhily commented on a change in pull request #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
syhily commented on a change in pull request #18816:
URL: https://github.com/apache/flink/pull/18816#discussion_r840254164
##########
File path: flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/groups/UnregisteredMetricsGroup.java
##########
@@ -178,4 +182,35 @@ public T getValue() {
return null;
}
}
+
+ private static class UnregisteredSinkWriterMetricGroup extends UnregisteredMetricsGroup
Review comment:
Why we need this implementation?
--
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 #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18816:
URL: https://github.com/apache/flink/pull/18816#issuecomment-1042851613
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31751",
"triggerID" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"triggerType" : "PUSH"
}, {
"hash" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31787",
"triggerID" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"triggerType" : "PUSH"
}, {
"hash" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31924",
"triggerID" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6b17fb81597d3d67dcd2367659245cc01707fec2 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31787)
* b7832f41b7aed0a43855fc2a79df01783802b4b4 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31924)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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 #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18816:
URL: https://github.com/apache/flink/pull/18816#issuecomment-1042851613
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31751",
"triggerID" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"triggerType" : "PUSH"
}, {
"hash" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31787",
"triggerID" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"triggerType" : "PUSH"
}, {
"hash" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31924",
"triggerID" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"triggerType" : "PUSH"
}, {
"hash" : "271648472e6f8e46df9945f096666a65d34f762c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31949",
"triggerID" : "271648472e6f8e46df9945f096666a65d34f762c",
"triggerType" : "PUSH"
}, {
"hash" : "26366818f1c6a7b29ddb063414d1b3a2eaf34408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33065",
"triggerID" : "26366818f1c6a7b29ddb063414d1b3a2eaf34408",
"triggerType" : "PUSH"
}, {
"hash" : "acd3f3dda40c6a2f6de81cdc2e0b20637468d141",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33142",
"triggerID" : "acd3f3dda40c6a2f6de81cdc2e0b20637468d141",
"triggerType" : "PUSH"
}, {
"hash" : "998140342e15a42f1272a780b9debcd53f04bf06",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33187",
"triggerID" : "998140342e15a42f1272a780b9debcd53f04bf06",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* acd3f3dda40c6a2f6de81cdc2e0b20637468d141 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33142)
* 998140342e15a42f1272a780b9debcd53f04bf06 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33187)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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 #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18816:
URL: https://github.com/apache/flink/pull/18816#issuecomment-1042851613
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31751",
"triggerID" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"triggerType" : "PUSH"
}, {
"hash" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31787",
"triggerID" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"triggerType" : "PUSH"
}, {
"hash" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31924",
"triggerID" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"triggerType" : "PUSH"
}, {
"hash" : "271648472e6f8e46df9945f096666a65d34f762c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "271648472e6f8e46df9945f096666a65d34f762c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b7832f41b7aed0a43855fc2a79df01783802b4b4 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31924)
* 271648472e6f8e46df9945f096666a65d34f762c UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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 #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18816:
URL: https://github.com/apache/flink/pull/18816#issuecomment-1042851613
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31751",
"triggerID" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"triggerType" : "PUSH"
}, {
"hash" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31787",
"triggerID" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"triggerType" : "PUSH"
}, {
"hash" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31924",
"triggerID" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"triggerType" : "PUSH"
}, {
"hash" : "271648472e6f8e46df9945f096666a65d34f762c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31949",
"triggerID" : "271648472e6f8e46df9945f096666a65d34f762c",
"triggerType" : "PUSH"
}, {
"hash" : "26366818f1c6a7b29ddb063414d1b3a2eaf34408",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33065",
"triggerID" : "26366818f1c6a7b29ddb063414d1b3a2eaf34408",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 26366818f1c6a7b29ddb063414d1b3a2eaf34408 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33065)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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 #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18816:
URL: https://github.com/apache/flink/pull/18816#issuecomment-1042851613
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31751",
"triggerID" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"triggerType" : "PUSH"
}, {
"hash" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31787",
"triggerID" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"triggerType" : "PUSH"
}, {
"hash" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31924",
"triggerID" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"triggerType" : "PUSH"
}, {
"hash" : "271648472e6f8e46df9945f096666a65d34f762c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31949",
"triggerID" : "271648472e6f8e46df9945f096666a65d34f762c",
"triggerType" : "PUSH"
}, {
"hash" : "26366818f1c6a7b29ddb063414d1b3a2eaf34408",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33065",
"triggerID" : "26366818f1c6a7b29ddb063414d1b3a2eaf34408",
"triggerType" : "PUSH"
}, {
"hash" : "acd3f3dda40c6a2f6de81cdc2e0b20637468d141",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33142",
"triggerID" : "acd3f3dda40c6a2f6de81cdc2e0b20637468d141",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 26366818f1c6a7b29ddb063414d1b3a2eaf34408 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33065)
* acd3f3dda40c6a2f6de81cdc2e0b20637468d141 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33142)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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] imaffe commented on a change in pull request #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
imaffe commented on a change in pull request #18816:
URL: https://github.com/apache/flink/pull/18816#discussion_r811614446
##########
File path: flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriterTest.java
##########
@@ -44,32 +47,198 @@
import org.apache.flink.util.UserCodeClassLoader;
import org.apache.pulsar.client.api.transaction.TransactionCoordinatorClient;
+import org.assertj.core.data.Offset;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
import java.util.Collection;
import java.util.OptionalLong;
+import java.util.concurrent.TimeUnit;
import static java.util.Collections.singletonList;
import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
+import static org.apache.flink.connector.base.DeliveryGuarantee.AT_LEAST_ONCE;
import static org.apache.flink.connector.base.DeliveryGuarantee.EXACTLY_ONCE;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_STATS_INTERVAL_SECONDS;
+import static org.apache.flink.connector.pulsar.sink.PulsarSinkOptions.PULSAR_PRODUCER_NAME;
import static org.apache.flink.connector.pulsar.sink.PulsarSinkOptions.PULSAR_WRITE_SCHEMA_EVOLUTION;
+import static org.apache.flink.connector.pulsar.sink.writer.metrics.PulsarProducerMetricsRegister.PRODUCER_UPDATE_POLLING_INTERVAL_MILLIS;
import static org.apache.flink.connector.pulsar.sink.writer.serializer.PulsarSerializationSchema.pulsarSchema;
+import static org.apache.flink.shaded.guava30.com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
import static org.apache.pulsar.client.api.Schema.STRING;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
/** Unit tests for {@link PulsarWriter}. */
class PulsarWriterTest extends PulsarTestSuiteBase {
-
private static final SinkWriter.Context CONTEXT = new MockSinkWriterContext();
+ private static final String DEFAULT_TEST_PRODUCER_NAME = "test-producer";
+ private static final String EXPECTED_ALL_PRODUCER_METRIC_NAME = "PulsarSink.numAcksReceived";
+ private static final String EXPECTED_PER_PRODUCER_METRIC_PREFIX = "PulsarSink.producer";
+ private static final String EXPECTED_PER_PRODUCER_METRIC_NAME = "sendLatency99Pct";
+ private static final String GLOBAL_MAX_LATENCY_METRIC_NAME = "PulsarSink.sendLatencyMax";
+ private static final String CURRENT_SEND_TIME_METRIC_NAME = "currentSendTime";
+ private static final String NUM_ACKS_RECEIVED_METRIC_NAME = "PulsarSink.numAcksReceived";
+ private static final long TEST_PULSAR_STATS_INTERVAL_SECONDS = 1L;
+
+ private MetricListener metricListener;
+ private TestProcessingTimeService timeService;
+ private MockInitContext initContext;
+
+ @BeforeEach
+ void setup() throws Exception {
+ metricListener = new MetricListener();
+ timeService = new TestProcessingTimeService();
+ timeService.setCurrentTime(0L);
+ initContext = new MockInitContext(metricListener, timeService);
+ }
@Test
void writeMessageWithGuarantee() throws Exception {
writeMessageWithoutGuarantee(EXACTLY_ONCE);
}
+ @Test
+ void metricsPresentAfterWriterCreated() throws Exception {
+ String topic = randomAlphabetic(10);
+ operator().createTopic(topic, 8);
+
+ try (final PulsarWriter<String> ignored = createWriter(topic, initContext)) {
+ assertThat(metricListener.getCounter(EXPECTED_ALL_PRODUCER_METRIC_NAME)).isPresent();
+ }
+ }
+
+ @Test
+ void perProducerMetricsPresentAfterMessageWritten() throws Exception {
+ String topic = randomAlphabetic(10);
+ operator().createTopic(topic, 8);
+
+ try (final PulsarWriter<String> writer = createWriter(topic, initContext)) {
+ String message = randomAlphabetic(10);
+ sendMessages(writer, 1);
+ // advance timer to update the producers
+ timeService.advance(PRODUCER_UPDATE_POLLING_INTERVAL_MILLIS + 1000);
+ assertThat(metricListener.getGauge(perProducerMetricName(topic, 0))).isPresent();
+
+ // send another message and advance timer
+ sendMessages(writer, 1000);
Review comment:
The 1000 here is a magic number, it should be equal to the switch size of round robin router
--
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] imaffe commented on a change in pull request #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
imaffe commented on a change in pull request #18816:
URL: https://github.com/apache/flink/pull/18816#discussion_r810455595
##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java
##########
@@ -270,5 +287,40 @@ public void flush(boolean endOfInput) throws IOException, InterruptedException {
public void close() throws Exception {
// Close all the resources and throw the exception at last.
closeAll(metadataListener, producerRegister);
+ closed = true;
+ }
+
+ private void setupFlinkMetrics() {
+ lastMetricUpdateTimestamp = timeService.getCurrentProcessingTime();
+ registerGlobalGauges();
+ registerMetricUpdateTimer();
+ }
+
+ /**
+ * Producer is lazy initialized in Pulsar Sink. Some metrics (currently only some gauges) can
+ * only be set after creating the producer. Those gauges are per producer wise. Some other
+ * metrics is based aggregation of all producers or they are based on sink implementations so
+ * they are called global gauges.
+ */
+ private void registerGlobalGauges() {
+ pulsarSinkWriterMetrics.setCurrentSendTimeGauge(this::getCurrentSendTime);
+ producerRegister.registerMaxSendLatencyGauges();
+ }
+
+ private void registerMetricUpdateTimer() {
Review comment:
Move these timer tasks to a dedicated 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
[GitHub] [flink] flinkbot commented on pull request #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #18816:
URL: https://github.com/apache/flink/pull/18816#issuecomment-1042851613
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0229cbdda1c15c851c2b594b721e388b9b0b7e5f UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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] syhily commented on a change in pull request #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
syhily commented on a change in pull request #18816:
URL: https://github.com/apache/flink/pull/18816#discussion_r827959064
##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/metrics/PulsarSinkWriterMetricUtils.java
##########
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.sink.writer.metrics;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
+
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerStats;
+import org.apache.pulsar.common.schema.SchemaInfo;
+
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/** Util class to provide monitor metrics methods to Sink Writer. */
+@Internal
+public class PulsarSinkWriterMetricUtils {
Review comment:
This class name makes me feel confused. The `utils` should be some class that only provided static methods.
--
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 #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18816:
URL: https://github.com/apache/flink/pull/18816#issuecomment-1042851613
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31751",
"triggerID" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"triggerType" : "PUSH"
}, {
"hash" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31787",
"triggerID" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"triggerType" : "PUSH"
}, {
"hash" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31924",
"triggerID" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"triggerType" : "PUSH"
}, {
"hash" : "271648472e6f8e46df9945f096666a65d34f762c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31949",
"triggerID" : "271648472e6f8e46df9945f096666a65d34f762c",
"triggerType" : "PUSH"
}, {
"hash" : "26366818f1c6a7b29ddb063414d1b3a2eaf34408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33065",
"triggerID" : "26366818f1c6a7b29ddb063414d1b3a2eaf34408",
"triggerType" : "PUSH"
}, {
"hash" : "acd3f3dda40c6a2f6de81cdc2e0b20637468d141",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33142",
"triggerID" : "acd3f3dda40c6a2f6de81cdc2e0b20637468d141",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* acd3f3dda40c6a2f6de81cdc2e0b20637468d141 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33142)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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] imaffe commented on a change in pull request #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
imaffe commented on a change in pull request #18816:
URL: https://github.com/apache/flink/pull/18816#discussion_r808982242
##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/TopicProducerRegister.java
##########
@@ -152,7 +156,7 @@ public void close() throws IOException {
builder.topic(topic);
Producer<T> producer = sneakyClient(builder::create);
producers.put(schemaInfo, producer);
-
+ pulsarSinkWriterMetrics.registerSingleProducerGauges(producer);
Review comment:
We can add producer metrics periodically in the timeService or we can add the metrics register logic here. Not sure which one is better
--
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 #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18816:
URL: https://github.com/apache/flink/pull/18816#issuecomment-1042851613
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31751",
"triggerID" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"triggerType" : "PUSH"
}, {
"hash" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31787",
"triggerID" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"triggerType" : "PUSH"
}, {
"hash" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31924",
"triggerID" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"triggerType" : "PUSH"
}, {
"hash" : "271648472e6f8e46df9945f096666a65d34f762c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31949",
"triggerID" : "271648472e6f8e46df9945f096666a65d34f762c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 271648472e6f8e46df9945f096666a65d34f762c Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31949)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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] imaffe commented on a change in pull request #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
imaffe commented on a change in pull request #18816:
URL: https://github.com/apache/flink/pull/18816#discussion_r840379481
##########
File path: flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/groups/UnregisteredMetricsGroup.java
##########
@@ -178,4 +182,35 @@ public T getValue() {
return null;
}
}
+
+ private static class UnregisteredSinkWriterMetricGroup extends UnregisteredMetricsGroup
Review comment:
Because I didn't find a good way to mock SinkWriterMetricGroup, and I saw there are other UnregisterdXXXMetricGroup() used in tests, so I decided to create one for sink.
--
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] imaffe commented on a change in pull request #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
imaffe commented on a change in pull request #18816:
URL: https://github.com/apache/flink/pull/18816#discussion_r840379481
##########
File path: flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/groups/UnregisteredMetricsGroup.java
##########
@@ -178,4 +182,35 @@ public T getValue() {
return null;
}
}
+
+ private static class UnregisteredSinkWriterMetricGroup extends UnregisteredMetricsGroup
Review comment:
Because I didn't find a good way to mock SinkWriterMetricGroup, and I saw there are other UnregisterdXXXMetricGroup() used in Sink
--
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 #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #18816:
URL: https://github.com/apache/flink/pull/18816#issuecomment-1042844761
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 0229cbdda1c15c851c2b594b721e388b9b0b7e5f (Thu Feb 17 11:19:54 UTC 2022)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] imaffe commented on a change in pull request #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
imaffe commented on a change in pull request #18816:
URL: https://github.com/apache/flink/pull/18816#discussion_r809181383
##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/metrics/PulsarSinkWriterMetrics.java
##########
@@ -0,0 +1,169 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.sink.writer.metrics;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
+
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerStats;
+import org.apache.pulsar.common.schema.SchemaInfo;
+
+import java.util.Comparator;
+import java.util.Map;
+
+/** Util class to provide monitor metrics methods to Sink Writer. */
+@Internal
+public class PulsarSinkWriterMetrics {
+ public static final String PULSAR_SINK_METRIC_GROUP = "PulsarSink";
+ public static final String PRODUCER_SUBGROUP = "producer";
+ public static final String METRIC_PRODUCER_SEND_LATENCY_MAX = "sendLatencyMax";
+ public static final String METRIC_NUM_ACKS_RECEIVED = "numAcksReceived";
+ public static final String METRIC_SEND_LATENCY_50_PCT = "sendLatency50Pct";
+ public static final String METRIC_SEND_LATENCY_75_PCT = "sendLatency75Pct";
+ public static final String METRIC_SEND_LATENCY_95_PCT = "sendLatency95Pct";
+ public static final String METRIC_SEND_LATENCY_99_PCT = "sendLatency99Pct";
+ public static final String METRIC_SEND_LATENCY_999_PCT = "sendLatency999Pct";
+ public static final double INVALID_LATENCY = -1d;
+ public static final long INVALID_LAST_SEND_TIME = -1;
+
+ private final SinkWriterMetricGroup sinkWriterMetricGroup;
+ private final MetricGroup producerMetricGroup;
+
+ // Standard counter metrics
+ private long lastNumBytesOut = 0;
+ private long lastNumRecordsOut = 0;
+ private long lastNumRecordsOutErrors = 0;
+
+ // Pulsar Producer counter metrics
+ private final Counter numAcksReceived;
+ private long lastNumAcksReceived = 0;
+
+ public PulsarSinkWriterMetrics(SinkWriterMetricGroup sinkWriterMetricGroup) {
+ this.sinkWriterMetricGroup = sinkWriterMetricGroup;
+ this.producerMetricGroup = sinkWriterMetricGroup.addGroup(PULSAR_SINK_METRIC_GROUP);
+ this.numAcksReceived = producerMetricGroup.counter(METRIC_NUM_ACKS_RECEIVED);
+ }
+
+ public <T> void updateProducerStats(
+ Map<String, Map<SchemaInfo, Producer<?>>> producerRegister) {
+ long numBytesOut = 0;
+ long numRecordsOut = 0;
+ long numRecordsOutErrors = 0;
+ long numAcksReceived = 0;
+
+ for (Map<SchemaInfo, Producer<?>> producers : producerRegister.values()) {
+ for (Producer<?> producer : producers.values()) {
+ ProducerStats stats = producer.getStats();
Review comment:
Split the loops and refactor to use functions
--
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 #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18816:
URL: https://github.com/apache/flink/pull/18816#issuecomment-1042851613
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31751",
"triggerID" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"triggerType" : "PUSH"
}, {
"hash" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31787",
"triggerID" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6b17fb81597d3d67dcd2367659245cc01707fec2 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31787)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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] imaffe commented on a change in pull request #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
imaffe commented on a change in pull request #18816:
URL: https://github.com/apache/flink/pull/18816#discussion_r808976119
##########
File path: flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/TopicProducerRegisterTest.java
##########
@@ -76,7 +76,7 @@ void noneAndAtLeastOnceWouldNotCreateTransaction(DeliveryGuarantee deliveryGuara
operator().createTopic(topic, 8);
SinkConfiguration configuration = sinkConfiguration(deliveryGuarantee);
- TopicProducerRegister register = new TopicProducerRegister(configuration);
+ TopicProducerRegister register = new TopicProducerRegister(configuration, null);
Review comment:
should not use null. Will throw NPE
--
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] imaffe commented on a change in pull request #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
imaffe commented on a change in pull request #18816:
URL: https://github.com/apache/flink/pull/18816#discussion_r808976119
##########
File path: flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/TopicProducerRegisterTest.java
##########
@@ -76,7 +76,7 @@ void noneAndAtLeastOnceWouldNotCreateTransaction(DeliveryGuarantee deliveryGuara
operator().createTopic(topic, 8);
SinkConfiguration configuration = sinkConfiguration(deliveryGuarantee);
- TopicProducerRegister register = new TopicProducerRegister(configuration);
+ TopicProducerRegister register = new TopicProducerRegister(configuration, null);
Review comment:
should not use null. Will throw NPE, use UnregisteredMetricsGroup
--
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] imaffe commented on a change in pull request #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
imaffe commented on a change in pull request #18816:
URL: https://github.com/apache/flink/pull/18816#discussion_r828062932
##########
File path: flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/metrics/PulsarSinkWriterMetricUtils.java
##########
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.sink.writer.metrics;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
+
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerStats;
+import org.apache.pulsar.common.schema.SchemaInfo;
+
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/** Util class to provide monitor metrics methods to Sink Writer. */
+@Internal
+public class PulsarSinkWriterMetricUtils {
Review comment:
Yeah I agree.
--
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 #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18816:
URL: https://github.com/apache/flink/pull/18816#issuecomment-1042851613
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31751",
"triggerID" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"triggerType" : "PUSH"
}, {
"hash" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31787",
"triggerID" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"triggerType" : "PUSH"
}, {
"hash" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31924",
"triggerID" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"triggerType" : "PUSH"
}, {
"hash" : "271648472e6f8e46df9945f096666a65d34f762c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31949",
"triggerID" : "271648472e6f8e46df9945f096666a65d34f762c",
"triggerType" : "PUSH"
}, {
"hash" : "26366818f1c6a7b29ddb063414d1b3a2eaf34408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33065",
"triggerID" : "26366818f1c6a7b29ddb063414d1b3a2eaf34408",
"triggerType" : "PUSH"
}, {
"hash" : "acd3f3dda40c6a2f6de81cdc2e0b20637468d141",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33142",
"triggerID" : "acd3f3dda40c6a2f6de81cdc2e0b20637468d141",
"triggerType" : "PUSH"
}, {
"hash" : "998140342e15a42f1272a780b9debcd53f04bf06",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33187",
"triggerID" : "998140342e15a42f1272a780b9debcd53f04bf06",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* acd3f3dda40c6a2f6de81cdc2e0b20637468d141 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33142)
* 998140342e15a42f1272a780b9debcd53f04bf06 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33187)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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 #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18816:
URL: https://github.com/apache/flink/pull/18816#issuecomment-1042851613
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31751",
"triggerID" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"triggerType" : "PUSH"
}, {
"hash" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31787",
"triggerID" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"triggerType" : "PUSH"
}, {
"hash" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31924",
"triggerID" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"triggerType" : "PUSH"
}, {
"hash" : "271648472e6f8e46df9945f096666a65d34f762c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31949",
"triggerID" : "271648472e6f8e46df9945f096666a65d34f762c",
"triggerType" : "PUSH"
}, {
"hash" : "26366818f1c6a7b29ddb063414d1b3a2eaf34408",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33065",
"triggerID" : "26366818f1c6a7b29ddb063414d1b3a2eaf34408",
"triggerType" : "PUSH"
}, {
"hash" : "acd3f3dda40c6a2f6de81cdc2e0b20637468d141",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "acd3f3dda40c6a2f6de81cdc2e0b20637468d141",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 26366818f1c6a7b29ddb063414d1b3a2eaf34408 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33065)
* acd3f3dda40c6a2f6de81cdc2e0b20637468d141 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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 #18816: [FLINK-26027][pulsar] add sink metrics to pulsar sink
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18816:
URL: https://github.com/apache/flink/pull/18816#issuecomment-1042851613
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31751",
"triggerID" : "0229cbdda1c15c851c2b594b721e388b9b0b7e5f",
"triggerType" : "PUSH"
}, {
"hash" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31787",
"triggerID" : "6b17fb81597d3d67dcd2367659245cc01707fec2",
"triggerType" : "PUSH"
}, {
"hash" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31924",
"triggerID" : "b7832f41b7aed0a43855fc2a79df01783802b4b4",
"triggerType" : "PUSH"
}, {
"hash" : "271648472e6f8e46df9945f096666a65d34f762c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=31949",
"triggerID" : "271648472e6f8e46df9945f096666a65d34f762c",
"triggerType" : "PUSH"
}, {
"hash" : "26366818f1c6a7b29ddb063414d1b3a2eaf34408",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33065",
"triggerID" : "26366818f1c6a7b29ddb063414d1b3a2eaf34408",
"triggerType" : "PUSH"
}, {
"hash" : "acd3f3dda40c6a2f6de81cdc2e0b20637468d141",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33142",
"triggerID" : "acd3f3dda40c6a2f6de81cdc2e0b20637468d141",
"triggerType" : "PUSH"
}, {
"hash" : "998140342e15a42f1272a780b9debcd53f04bf06",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33187",
"triggerID" : "998140342e15a42f1272a780b9debcd53f04bf06",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 998140342e15a42f1272a780b9debcd53f04bf06 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=33187)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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