[FLINK-6998][Kafka Connector] Add kafka offset commit metrics in cons… #4187

zhenzhongxu · 2017-06-26T22:52:09Z

add "kafkaCommitsSucceeded" and "kafkaCommitsFailed" metrics in KafkaConsumerThread class.

zentol · 2017-06-27T10:05:08Z

Please resolve the checkstyle violations:


[ERROR] src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThread.java:[96] (javadoc) JavadocStyle: First sentence should end with a period.
[ERROR] src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThread.java:[99] (javadoc) JavadocStyle: First sentence should end with a period.

tzulitai · 2017-06-27T10:19:15Z

Hi @zhenzhongxu, thanks for the PR. I like the idea of exposing these metrics.

Regarding the metric naming:
I think we also expose metrics that are directly available through the Kafka client.
Apart from this PR, there was also previous discussion about adding other Flink-specific metrics to the connector (e.g., offset lag since last checkpoint).
I wonder whether or not we should regulate the naming scheme for these Flink-specific metrics, to set them apart from Kafka's provided metrics.

Regarding the implementation:
This current solution only exposes the metric for Kafka 09 and 010.
Ideally, this should be added to the FlinkKafkaConsumerBase, so that we don't end up doing the same thing in future versions. It would be a bit more work, such as providing a callback to AbstractFetcher.commitInternalOffsetsToKafka, but then it would be properly implemented.

What do you think?

tzulitai · 2017-06-27T10:20:32Z

docs/monitoring/metrics.md

+    <tr>
+      <th rowspan="1">Slot/Consumer</th>
+      <td>kafkaCommitsSucceeded</td>
+      <td>Kafka offset commit success count if Kafka commit is turned on.</td>


I would make this statement more strict, in that the metric only exists if commit is turned on AND checkpointing is enabled. The added metrics would not appear if only Kafka's periodic offset committing is turned on.

Sure, thanks for the feedback. Will update.a

tzulitai · 2017-06-27T10:20:39Z

docs/monitoring/metrics.md

+    <tr>
+       <th rowspan="1">Slot/Consumer</th>
+       <td>kafkaCommitsFailed</td>
+       <td>Kafka offset commit failure count if Kafka commit is turned on.</td>


Sure, thanks for the feedback. Will update.a

tzulitai · 2017-06-27T14:11:13Z

.../src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThread.java

@@ -119,6 +126,9 @@ public KafkaConsumerThread(

 		this.nextOffsetsToCommit = new AtomicReference<>();
 		this.running = true;
+
+		this.successfulCommits = kafkaMetricGroup.counter("kafkaCommitsSucceeded");
+		this.failedCommits = kafkaMetricGroup.counter("kafkaCommitsFailed");


Please see my comment regarding the naming of the metrics.

zhenzhongxu · 2017-06-27T22:43:03Z

@tzulitai

Regarding the metric naming:
Any suggestions on naming conventions for these flink specific metrics? How do you like 'kafkaconnector-commits-succeeded' (component-metric-name) as an example? I personally like hyphen seperators better than camel case for metric names.
I'll not include the other proposed metric in this PR just for the sake of simplicity. I also have some opinions on "offset lag" metric, I think this particular metric is more useful when some external entity
perform the monitoring (difference of committed offset vs log head), especially in failure situations.

Regarding the implementation:
Thanks for the feedback. I'll explore the more proper implementation suggested, I'll get back to you with a solution or question.

zentol · 2017-06-28T09:12:34Z

I don't think it matters to much whether a metric is completely measured by Flink or just forwarded from kafka classes, so i would not add "kafka" to the metric name. Having "kafka" in the name also introduces an inherent redundancy, since the scope for, say the KafkaConsumerThread, already contains "KafkaConsumer".

zhenzhongxu · 2017-06-28T22:30:16Z

How about just "commits-succeeded" and "commits-failed" as metric names.

tzulitai · 2017-06-29T08:24:21Z

I agree. "commits-succeeded" and "commits-failed" seems good!

tzulitai · 2017-06-29T08:25:33Z

I'll not include the other proposed metric in this PR just for the sake of simplicity. I also have some opinions on "offset lag" metric, I think this particular metric is more useful when some external entity
perform the monitoring (difference of committed offset vs log head), especially in failure situations.

Yes, lets keep that apart from this PR. There is also a JIRA for exactly this feature: https://issues.apache.org/jira/browse/FLINK-6109

zentol · 2017-06-29T09:17:20Z

Please follow the camel-case pattern that we use for other metrics.

tzulitai · 2017-07-05T06:09:57Z

Hi @zhenzhongxu, could you rebase on to the latest master? Currently the PR contains commits unrelated to the change.

zhenzhongxu · 2017-07-05T18:48:24Z

@tzulitai rebase done.

GJL · 2017-07-08T14:38:38Z

...a-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java

@@ -104,7 +106,7 @@
 	//  configuration state, set on the client relevant for all subtasks
 	// ------------------------------------------------------------------------

-	/** Describes whether we are discovering partitions for fixed topics or a topic pattern. */
+	/** Descrnterbes whether we are discovering partitions for fixed topics or a topic pattern. */


This is probably an unintended change.

GJL · 2017-07-08T14:39:15Z

...a-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java

@@ -184,6 +186,18 @@
 	/** Flag indicating whether the consumer is still running. */
 	private volatile boolean running = true;

+    // ------------------------------------------------------------------------


The indentation is off by 1 space.

zhenzhongxu · 2017-07-13T23:21:22Z

@tzulitai seems the last CI pipeline failed because of stability issues, how can I trigger another build without making a commit?

zentol · 2017-07-14T07:29:25Z

docs/monitoring/metrics.md

+  </thead>
+  <tbody>
+    <tr>
+      <th rowspan="1">Slot/Consumer</th>


This documentation is inconsistent with the rest.

Scope should be "Operator", and you should add an additional "Infix" column which contains the names of the metric groups you are creating, concatenated with a period.

zentol · 2017-07-14T07:30:49Z

...a-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java

+			@Override
+			public void onComplete(Exception exception) {
+				if (exception == null) {
+					successfulCommits.inc();


If the callbacks are executed by another thread the result will be inaccurate, as the default counter implementation is not thread-safe.

This is only invoked during checkpoint complete notification. There is no thread race condition.

I would also like to raise a thread-safety issue here.

Currently, since there's always only one pending offset commit in Kafka 09+, and Kafka08 commits in a blocking call, there will be no race condition in incrementing the counters. However, changing these implementations in subclasses (perhaps in the future) can easily introduce race conditions here.

At the very least, we probably should add a notice about thread-safety contract in the Javadoc of commitInternalOffsetsToKafka.

I think I'll add a detailed javadoc to describe the current unprotected implementation. I am hesitant to add in lock protection because higher level abstraction currently guarantees no concurrent commit at the moment.

zentol · 2017-07-14T07:33:02Z

...src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaCommitCallback.java

+package org.apache.flink.streaming.connectors.kafka.internals;
+
+/**
+ * A callback interface that the user can implement to trigger custom actions when a commit request completes.


Is this callback actually exposed to users?

I updated the wording, this is only exposed to the kafka source operator (invoked upon checkpoint compete), not directly to users. Hence, there is no thread race condition either.

tzulitai

Thanks for the updates @zhenzhongxu.
Overall I like the changes so far! I have some comments inline.

tzulitai · 2017-07-17T05:57:04Z

.../src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThread.java

+	 *
+	 * <p>Only one commit operation may be pending at any time. If the committing takes longer than
+	 * the frequency with which this method is called, then some commits may be skipped due to being
+	 * superseded  by newer ones.


Unnecessary empty space.

tzulitai · 2017-07-17T05:57:42Z

...a-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java

+	/** Counter for failed Kafka offset commits. */
+	private transient Counter failedCommits;
+
+	private transient KafkaCommitCallback offsetCommitCallback;


Can you include a Javadoc for this too? (for consistency)

Sounds fair, I'll include a javadoc as well a notice about the thread-safety contract as you suggested.

tzulitai · 2017-07-17T06:00:53Z

...a-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java

+			@Override
+			public void onComplete(Exception exception) {
+				if (exception == null) {
+					successfulCommits.inc();


I would also like to raise a thread-safety issue here.

Currently, since there's always only one pending offset commit in Kafka 09+, and Kafka08 commits in a blocking call, there will be no race condition in incrementing the counters. However, changing these implementations in subclasses (perhaps in the future) can easily introduce race conditions here.

At the very least, we probably should add a notice about thread-safety contract in the Javadoc of commitInternalOffsetsToKafka.

tzulitai · 2017-07-17T06:02:52Z

...src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaCommitCallback.java

+/**
+ * A callback interface that the user can implement to trigger custom actions when a commit request completes.
+ */
+public interface KafkaCommitCallback {


Not entirely sure of the need of an interface here. Do we allow "version-specific subclasses" to have their own KafkaCommitCallback implementation?

If not, then this makes the usage slightly confusing.

zentol · 2017-07-18T12:58:18Z

docs/monitoring/metrics.md

@@ -896,6 +896,28 @@ Thus, in order to infer the metric identifier:
  </tbody>
 </table>

+#### Connector:


Let's name this heading "Connectors", and add a new sub-heading specifically for kafka.

ok make sense

zhenzhongxu · 2017-07-20T22:30:17Z

@tzulitai all tests passing now. let me know if this looks ok now, also let me know if you want me to go ahead squash all commits.

StephanEwen · 2017-07-26T18:48:11Z

...src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaCommitCallback.java

+ * A callback interface that the source operator can implement to trigger custom actions when a commit request completes,
+ * which should normally be triggered from checkpoint complete event.
+ */
+public interface KafkaCommitCallback {


I think it would be nice to have two methods here: onSuccess() and onException(...).
Or does this have to be a SAM interface so you can use lambdas?

Having the two methods would make the om success case more clear...

StephanEwen · 2017-07-26T18:49:09Z

...src/main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaCommitCallback.java

+	 *
+	 * @param exception The exception thrown during processing of the request, or null if the commit completed successfully
+	 */
+	void onComplete(Exception exception);


Most other exception handlers take a Throwable. Would it make sense to do that here as well?

…umer callback 3CR feedback;make code change backward compatible with test cases;fix an earlier merge issue add sub-heading for kafka connectors in metrics.md

…on KafkaCommitCallback interface

tzulitai

I think this is good to merge now, +1.
I have some final cosmetic stuff I would like to be changed, will address them myself when merging!
Thanks for the contribution @zhenzhongxu.

tzulitai · 2017-07-28T10:17:01Z

.../src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThread.java

@@ -308,17 +311,33 @@ public void shutdown() {
 	 *
 	 * <p>Only one commit operation may be pending at any time. If the committing takes longer than
 	 * the frequency with which this method is called, then some commits may be skipped due to being
-	 * superseded  by newer ones.
+	 * superseded by newer ones.
 	 *
 	 * @param offsetsToCommit The offsets to commit
 	 */
 	public void setOffsetsToCommit(Map<TopicPartition, OffsetAndMetadata> offsetsToCommit) {


I don't think we need this variant anymore. The only user of the method is Kafka09Fetcher, anyways.

tzulitai · 2017-07-28T10:19:00Z

...-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/Kafka08Fetcher.java

 		ZookeeperOffsetHandler zkHandler = this.zookeeperOffsetHandler;
 		if (zkHandler != null) {
 			try {
 				// the ZK handler takes care of incrementing the offsets by 1 before committing
 				zkHandler.prepareAndCommitOffsets(offsets);
+				if (commitCallback != null) {


I would actually like to remove these null checks, and have the contract that a callback will always be provided with @Nonnull annotation.

AFAIK, the only reason we need these null checks is that the tests, for simplicity, provide a null as the callback. IMO, it isn't a good practice to have logic in the main code just to satisfy testing shortcuts.

tzulitai · 2017-07-28T10:19:23Z

...-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/Kafka08Fetcher.java

 			}
 			catch (Exception e) {
 				if (running) {
+					if (commitCallback != null) {


See my comment above. Would like to remove these null checks.

tzulitai · 2017-07-28T10:19:37Z

.../src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThread.java

@@ -482,6 +501,12 @@ public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception

 			if (ex != null) {
 				log.warn("Committing offsets to Kafka failed. This does not compromise Flink's checkpoints.", ex);
+				if (callerCommitCallback != null) {


See my comment above. Would like to remove these null checks.

tzulitai · 2017-07-28T10:19:43Z

.../src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThread.java

+					callerCommitCallback.onException(ex);
+				}
+			}
+			else if (callerCommitCallback != null) {


See my comment above. Would like to remove these null checks.

…ommitCallbacks This closes apache#4187.

tzulitai · 2017-07-28T13:34:10Z

Merging ..

…ommitCallbacks This closes apache#4187.

tzulitai requested changes Jun 27, 2017

View reviewed changes

zhenzhongxu force-pushed the FLINK-6998 branch from e429332 to 47170e2 Compare July 5, 2017 18:42

GJL reviewed Jul 8, 2017

View reviewed changes

zentol reviewed Jul 14, 2017

View reviewed changes

tzulitai reviewed Jul 17, 2017

View reviewed changes

zentol reviewed Jul 18, 2017

View reviewed changes

StephanEwen reviewed Jul 26, 2017

View reviewed changes

zhenzhongxu force-pushed the FLINK-6998 branch from 91bf459 to 2aea137 Compare July 27, 2017 18:15

Zhenzhong Xu added 2 commits July 27, 2017 16:21

[FLINK-6998][Kafka Connector] Add kafka offset commit metrics in cons…

6119c5a

…umer callback 3CR feedback;make code change backward compatible with test cases;fix an earlier merge issue add sub-heading for kafka connectors in metrics.md

[FLINK-6998] Distinguish between onException and OnSuccess callbacks …

17ba7b5

…on KafkaCommitCallback interface

zhenzhongxu force-pushed the FLINK-6998 branch from 0a13933 to 17ba7b5 Compare July 27, 2017 23:22

tzulitai approved these changes Jul 28, 2017

View reviewed changes

tzulitai added a commit to tzulitai/flink that referenced this pull request Jul 28, 2017

[FLINK-6998] [kafka] Remove unfruitful null checks on provided KafkaC…

2d65c7e

…ommitCallbacks This closes apache#4187.

asfgit closed this in 1ded2d8 Jul 28, 2017

tzulitai added a commit to tzulitai/flink that referenced this pull request Jul 28, 2017

[FLINK-6998] [kafka] Remove unfruitful null checks on provided KafkaC…

4382464

…ommitCallbacks This closes apache#4187.

rmetzger added the component=Connectors/Kafka label Mar 14, 2019

[FLINK-6998][Kafka Connector] Add kafka offset commit metrics in cons… #4187

[FLINK-6998][Kafka Connector] Add kafka offset commit metrics in cons… #4187

Conversation

zhenzhongxu commented Jun 26, 2017

zentol commented Jun 27, 2017

tzulitai commented Jun 27, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhenzhongxu commented Jun 27, 2017

zentol commented Jun 28, 2017 • edited

zhenzhongxu commented Jun 28, 2017

tzulitai commented Jun 29, 2017

tzulitai commented Jun 29, 2017

zentol commented Jun 29, 2017

tzulitai commented Jul 5, 2017

zhenzhongxu commented Jul 5, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhenzhongxu commented Jul 13, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tzulitai left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhenzhongxu commented Jul 20, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tzulitai left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tzulitai commented Jul 28, 2017

tzulitai commented Jun 27, 2017 •

edited

zentol commented Jun 28, 2017 •

edited