[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16652298#comment-16652298
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

zentol closed pull request #1157: [FLINK-2720][storm-compatibility]Add 
Storm-CountMetric for storm metrics
URL: https://github.com/apache/flink/pull/1157
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git 
a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/api/FlinkTopologyContext.java
 
b/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/api/FlinkTopologyContext.java
index a7616170d5b..5d20755339d 100644
--- 
a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/api/FlinkTopologyContext.java
+++ 
b/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/api/FlinkTopologyContext.java
@@ -26,6 +26,9 @@
 import backtype.storm.metric.api.ReducedMetric;
 import backtype.storm.state.ISubscribedState;
 import backtype.storm.task.TopologyContext;
+import org.apache.flink.api.common.accumulators.LongCounter;
+import org.apache.flink.stormcompatibility.api.metrics.FlinkCountMetric;
+import org.apache.flink.streaming.runtime.tasks.StreamingRuntimeContext;
 
 import java.util.Collection;
 import java.util.Map;
@@ -36,6 +39,8 @@
  */
 public class FlinkTopologyContext extends TopologyContext {
 
+   private StreamingRuntimeContext runtimeContext;
+
/**
 * Instantiates a new {@link FlinkTopologyContext} for a given Storm 
topology. The context object is instantiated
 * for each parallel task
@@ -53,6 +58,10 @@ public FlinkTopologyContext(final StormTopology topology, 
final Maphttp://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.stormcompatibility.api.metrics;
+
+import backtype.storm.metric.api.CountMetric;
+import org.apache.flink.api.common.accumulators.LongCounter;
+
+/**
+ * The wrapper for using {@link backtype.storm.metric.api.CountMetric}
+ */
+public class FlinkCountMetric extends CountMetric {
+
+   private LongCounter longCounter;
+
+   public FlinkCountMetric () {
+   }
+
+   public void setCounter (LongCounter longCounter) {
+   this.longCounter = longCounter;
+   }
+
+   @Override
+   public void incr() {
+   incrBy(1);
+   }
+
+   @Override
+   public void incrBy(long incrementBy) {
+   this.longCounter.add(incrementBy);
+   }
+
+   @Override
+   public Object getValueAndReset() {
+   long ret = this.longCounter.getLocalValue();
+   this.longCounter.resetLocal();
+   return ret;
+   }
+}
diff --git 
a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/wrappers/StormWrapperSetupHelper.java
 
b/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/wrappers/StormWrapperSetupHelper.java
index 75ab8e0d8c0..16bc4f53866 100644
--- 
a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/wrappers/StormWrapperSetupHelper.java
+++ 
b/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/wrappers/StormWrapperSetupHelper.java
@@ -104,7 +104,10 @@ public static TopologyContext 
convertToTopologyContext(final StreamingRuntimeCon
bolts.put(context.getTaskName(), new Bolt(null, 
common));
}
 
-   return new FlinkTopologyContext(new StormTopology(spoutSpecs, 
bolts, null), taskToComponents, taskId);
+   FlinkTopologyContext flinkContext = new 
FlinkTopologyContext(new StormTopology(spoutSpecs, bolts, null), 
taskToComponents, taskId);
+
+   flinkContext.setContext(context);
+   return flinkContext;
}
 
 }
diff --git 
a/flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/test/java/org/apache/flink/stormcompatibility/api/metrics/FlinkCountMetricTest.java
 

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16652297#comment-16652297
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

zentol commented on issue #1157: [FLINK-2720][storm-compatibility]Add 
Storm-CountMetric for storm metrics
URL: https://github.com/apache/flink/pull/1157#issuecomment-430367074
 
 
   subsumed by #3615.


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
>Priority: Major
>  Labels: pull-request-available
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16381923#comment-16381923
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user RalphSu commented on the issue:

https://github.com/apache/flink/pull/3615
  
update with latest master. Along fix the comments above.


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
>Priority: Major
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15953312#comment-15953312
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user zentol commented on a diff in the pull request:

https://github.com/apache/flink/pull/3615#discussion_r109393713
  
--- Diff: 
flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/metric/CounterMetricAdapter.java
 ---
@@ -0,0 +1,94 @@
+/*
+ * 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.storm.metric;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Metric;
+
+import backtype.storm.metric.api.CountMetric;
+import backtype.storm.metric.api.IMetric;
+
+/**
+ * An adapter to compose the counter metric between flink counter and 
storm counter metric.
+ * 
+ * @since Mar 5, 2017
+ *
+ */
+public class CounterMetricAdapter extends CountMetric implements IMetric, 
Counter, IMetricAdapter {
+
+/** the current count */
+// private long count;
+
+private Counter delegate;
+
+public CounterMetricAdapter(Counter counter) {
+this.delegate = counter;
+}
+
+// /Overide of storm CountMetric method to rely on current field 
'count' instead of '_value' in parent class.///
+
+public void incr() {
+delegate.inc();
+}
+
+public void incrBy(long incrementBy) {
+delegate.inc(incrementBy);
+}
+
+public Object getValueAndReset() {
--- End diff --

Why does this return Object and not long?


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15953306#comment-15953306
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user zentol commented on a diff in the pull request:

https://github.com/apache/flink/pull/3615#discussion_r109391296
  
--- Diff: 
flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/wrappers/FlinkTopologyContext.java
 ---
@@ -115,14 +126,16 @@ public ReducedMetric registerMetric(final String 
name, final IReducer combiner,
}
 
/**
-* Not supported by Flink.
-*
-* @throws UnsupportedOperationException
-*  at every invocation
+* @return - null when storm metric not supported by flink
 */
@Override
public  T registerMetric(final String name, final T 
metric, final int timeBucketSizeInSecs) {
-   throw new UnsupportedOperationException("Metrics are not 
supported by Flink");
+   IMetricAdapter adapter = 
MetricConvert.convertFlinkAdapter(name, metric, context);
+if (adapter == null) {
+LOG.info("Metric not supportted by Flink");
--- End diff --

typo: supportted -> supported


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15952916#comment-15952916
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user mjsax commented on the issue:

https://github.com/apache/flink/pull/3615
  
@RalphSu Do you have an JIRA id so we can assign the ticket to you?


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15952914#comment-15952914
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user mjsax commented on the issue:

https://github.com/apache/flink/pull/1157
  
@HuangWHWHW Can you please close this PR? @RalphSu is working on this now 
(cf #3615 ). Thanks.


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15952912#comment-15952912
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user mjsax commented on a diff in the pull request:

https://github.com/apache/flink/pull/3615#discussion_r109326653
  
--- Diff: 
flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/metric/IMetricAdapter.java
 ---
@@ -0,0 +1,34 @@
+/*
+ * 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.storm.metric;
+
+import org.apache.flink.metrics.Metric;
+import org.apache.storm.metric.api.IMetric;
+
+
+/**
+ * An adapter interface that returns storm and flink metric instance that 
back eath other.
+ * 
+ * @since Mar 25, 2017
+ *
+ */
+public interface IMetricAdapter {
--- End diff --

I don't understand the purpose of this interface. IMHO, we don't need it.


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15952908#comment-15952908
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user mjsax commented on a diff in the pull request:

https://github.com/apache/flink/pull/3615#discussion_r109326772
  
--- Diff: 
flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/metric/MultiCountMetricAdapter.java
 ---
@@ -0,0 +1,69 @@
+/*
+ * 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.storm.metric;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Metric;
+import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
+
+import backtype.storm.metric.api.MultiCountMetric;
+
+/**
+ * @since Mar 5, 2017
+ *
+ */
+public class MultiCountMetricAdapter extends MultiCountMetric implements 
Metric, IMetricAdapter {
+
+private Map _value = new HashMap<>();
--- End diff --

Nit: rename `_value` -> `counterPerKey`


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15952911#comment-15952911
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user mjsax commented on a diff in the pull request:

https://github.com/apache/flink/pull/3615#discussion_r109326571
  
--- Diff: 
flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/metric/CounterMetricAdapter.java
 ---
@@ -0,0 +1,94 @@
+/*
+ * 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.storm.metric;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Metric;
+
+import backtype.storm.metric.api.CountMetric;
+import backtype.storm.metric.api.IMetric;
+
+/**
+ * An adapter to compose the counter metric between flink counter and 
storm counter metric.
+ * 
+ * @since Mar 5, 2017
+ *
+ */
+public class CounterMetricAdapter extends CountMetric implements IMetric, 
Counter, IMetricAdapter {
+
+/** the current count */
+// private long count;
--- End diff --

Nit: remove this.


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15952913#comment-15952913
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user mjsax commented on a diff in the pull request:

https://github.com/apache/flink/pull/3615#discussion_r109326583
  
--- Diff: 
flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/metric/CounterMetricAdapter.java
 ---
@@ -0,0 +1,94 @@
+/*
+ * 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.storm.metric;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Metric;
+
+import backtype.storm.metric.api.CountMetric;
+import backtype.storm.metric.api.IMetric;
+
+/**
+ * An adapter to compose the counter metric between flink counter and 
storm counter metric.
+ * 
+ * @since Mar 5, 2017
+ *
+ */
+public class CounterMetricAdapter extends CountMetric implements IMetric, 
Counter, IMetricAdapter {
+
+/** the current count */
+// private long count;
+
+private Counter delegate;
+
+public CounterMetricAdapter(Counter counter) {
--- End diff --

nit: add `final` to all parameter (do this for all methods)


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15952909#comment-15952909
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user mjsax commented on a diff in the pull request:

https://github.com/apache/flink/pull/3615#discussion_r109326550
  
--- Diff: 
flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/metric/CounterMetricAdapter.java
 ---
@@ -0,0 +1,94 @@
+/*
+ * 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.storm.metric;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Metric;
+
+import backtype.storm.metric.api.CountMetric;
+import backtype.storm.metric.api.IMetric;
+
+/**
+ * An adapter to compose the counter metric between flink counter and 
storm counter metric.
+ * 
+ * @since Mar 5, 2017
+ *
+ */
+public class CounterMetricAdapter extends CountMetric implements IMetric, 
Counter, IMetricAdapter {
--- End diff --

Why does this class inherit from `Counter`? I seems to be sufficient that 
it as a member `Counter`.


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15952910#comment-15952910
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user mjsax commented on a diff in the pull request:

https://github.com/apache/flink/pull/3615#discussion_r109326716
  
--- Diff: 
flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/metric/MetricConvert.java
 ---
@@ -0,0 +1,49 @@
+/*
+ * 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.storm.metric;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
+import org.apache.storm.metric.api.CountMetric;
+import org.apache.storm.metric.api.IMetric;
+import org.apache.storm.metric.api.MultiCountMetric;
+
+/**
+ * @since Mar 25, 2017
+ *
+ */
+public class MetricConvert {
+
+   /**
+* 
+* @param name
+* @param originalMetric
+* @param context
+* @return
+*/
+   @SuppressWarnings("unchecked")
+   public static  IMetricAdapter 
convertFlinkAdapter(String name, T originalMetric,
+   StreamingRuntimeContext context) {
+   if (originalMetric instanceof CountMetric) {
+   Counter flinkc = context.getMetricGroup().counter(name);
+   return (IMetricAdapter) new 
CounterMetricAdapter(flinkc);
+   } else if (originalMetric instanceof MultiCountMetric) {
+   return (IMetricAdapter) new 
MultiCountMetricAdapter(context);
+   }
+   return null;
--- End diff --

I guess it's better to throw an exception here.


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15952907#comment-15952907
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user mjsax commented on a diff in the pull request:

https://github.com/apache/flink/pull/3615#discussion_r109326559
  
--- Diff: 
flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/metric/CounterMetricAdapter.java
 ---
@@ -0,0 +1,94 @@
+/*
+ * 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.storm.metric;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Metric;
+
+import backtype.storm.metric.api.CountMetric;
+import backtype.storm.metric.api.IMetric;
+
+/**
+ * An adapter to compose the counter metric between flink counter and 
storm counter metric.
+ * 
+ * @since Mar 5, 2017
--- End diff --

Nit: remove this line. (also in other classes)


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15943063#comment-15943063
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user greghogan commented on the issue:

https://github.com/apache/flink/pull/3615
  
Multiple PRs are added to one ticket but the commit header must reference 
the Jira ID. I think the comments here are sufficient so there is no need to 
close and open a new PR.


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15942564#comment-15942564
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user RalphSu commented on the issue:

https://github.com/apache/flink/pull/1157
  
I added https://github.com/apache/flink/pull/3615, welcome for comments.


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15898269#comment-15898269
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user mjsax commented on the issue:

https://github.com/apache/flink/pull/1157
  
@RalphSu We should wait for @HuangWHWHW to reply... If no response comes 
within 3 days you can take over. Also, the JIRA should get assigned to you for 
this case: https://issues.apache.org/jira/browse/FLINK-2720


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15896722#comment-15896722
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user RalphSu commented on the issue:

https://github.com/apache/flink/pull/1157
  
okey, would try file a PR on this soon


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15896579#comment-15896579
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user mjsax commented on the issue:

https://github.com/apache/flink/pull/1157
  
@HuangWHWHW Are you still working on this? @RalphSu Are you interested in 
talking this over?


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15896239#comment-15896239
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user RalphSu commented on the issue:

https://github.com/apache/flink/pull/1157
  
?


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15896238#comment-15896238
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user RalphSu commented on the issue:

https://github.com/apache/flink/pull/1157
  
Here it looks user need to change code to make metrics work


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15896237#comment-15896237
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user RalphSu commented on the issue:

https://github.com/apache/flink/pull/1157
  
Here it looks user need to change code to make metrics work


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15896234#comment-15896234
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user RalphSu commented on the issue:

https://github.com/apache/flink/pull/1157
  
Any progress on this one? Looks this has been stop for a while.


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15896231#comment-15896231
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user RalphSu commented on a diff in the pull request:

https://github.com/apache/flink/pull/1157#discussion_r104311095
  
--- Diff: 
flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/api/FlinkTopologyContext.java
 ---
@@ -120,7 +129,19 @@ public ReducedMetric registerMetric(final String name, 
final IReducer combiner,
@SuppressWarnings("unchecked")
@Override
public IMetric registerMetric(final String name, final IMetric metric, 
final int timeBucketSizeInSecs) {
-   throw new UnsupportedOperationException("Metrics are not 
supported by Flink");
+
+   // TODO: There is no use for timeBucketSizeInSecs yet.
+   if (metric instanceof FlinkCountMetric) {
--- End diff --

Here it looks user need to change code to make metrics work


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15896228#comment-15896228
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user RalphSu commented on a diff in the pull request:

https://github.com/apache/flink/pull/1157#discussion_r104311065
  
--- Diff: 
flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/api/FlinkTopologyContext.java
 ---
@@ -120,7 +129,19 @@ public ReducedMetric registerMetric(final String name, 
final IReducer combiner,
@SuppressWarnings("unchecked")
@Override
public IMetric registerMetric(final String name, final IMetric metric, 
final int timeBucketSizeInSecs) {
-   throw new UnsupportedOperationException("Metrics are not 
supported by Flink");
+
+   // TODO: There is no use for timeBucketSizeInSecs yet.
+   if (metric instanceof FlinkCountMetric) {
--- End diff --

Here it looks user need to change code to make metrics work


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15822025#comment-15822025
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user mjsax commented on the issue:

https://github.com/apache/flink/pull/1157
  
Did not have a chance to work on this either -- I am in a new job, too :)

You can pick this up again if you are still interested.


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15096485#comment-15096485
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user mjsax commented on the pull request:

https://github.com/apache/flink/pull/1157#issuecomment-171352040
  
As it seems you are not working on this any longer, I take over the JIRA. 
Please close this PR. Otherwise, I will close it the next days.


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Matthias J. Sax
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15013970#comment-15013970
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user mjsax commented on the pull request:

https://github.com/apache/flink/pull/1157#issuecomment-158127308
  
@HuangWHWHW What is the state of this PR? If you don't want to finish this, 
please close the PR (you can reopen a new PR any time later if you start to 
work on it again -- or are you still on it?)


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Huang Wei
> Fix For: 1.0.0
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=14982242#comment-14982242
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user mjsax commented on the pull request:

https://github.com/apache/flink/pull/1157#issuecomment-152474074
  
Hi. Are you still working on this? It's a couple of weeks since the last 
update.


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Huang Wei
> Fix For: 0.10
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=14960373#comment-14960373
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user HuangWHWHW commented on the pull request:

https://github.com/apache/flink/pull/1157#issuecomment-148657366
  
@mjsax
Ah, very sorry to keep you wait.
Due to the National Day, I was on the vocation last week and I get some 
hard jobs this week.
I'll rework this during this weekend and will reply to you next Monday.
:-)


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Huang Wei
> Fix For: 0.10
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=14960374#comment-14960374
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user mjsax commented on the pull request:

https://github.com/apache/flink/pull/1157#issuecomment-148658081
  
There is no need to rush. I was just curious. :)


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Huang Wei
> Fix For: 0.10
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=14958817#comment-14958817
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user mjsax commented on the pull request:

https://github.com/apache/flink/pull/1157#issuecomment-148370496
  
Any progress on this?


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Huang Wei
> Fix For: 0.10
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=14902221#comment-14902221
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user mjsax commented on the pull request:

https://github.com/apache/flink/pull/1157#issuecomment-142210957
  
Yes, but I don't understand the purpose of `FlinkCounMetric`... Flink 
should be able to deal with Storm metrics directly (without requiring the user 
to use a special Flink metric class).


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Huang Wei
> Fix For: 0.10
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=14901893#comment-14901893
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user HuangWHWHW commented on the pull request:

https://github.com/apache/flink/pull/1157#issuecomment-142176942
  
@mjsax 
Yes, I got the same example as you :D
And I try to add an example to show how to use this FlinkCountMetric by 
following the example.


> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Huang Wei
> Fix For: 0.10
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=14900723#comment-14900723
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

Github user mjsax commented on a diff in the pull request:

https://github.com/apache/flink/pull/1157#discussion_r39972364
  
--- Diff: 
flink-contrib/flink-storm-compatibility/flink-storm-compatibility-core/src/main/java/org/apache/flink/stormcompatibility/api/FlinkTopologyContext.java
 ---
@@ -53,6 +58,10 @@ public FlinkTopologyContext(final StormTopology 
topology, final Map Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Huang Wei
> Fix For: 0.10
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-2720) Add Storm-CountMetric in flink-stormcompatibility

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2720?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=14900686#comment-14900686
 ] 

ASF GitHub Bot commented on FLINK-2720:
---

GitHub user HuangWHWHW opened a pull request:

https://github.com/apache/flink/pull/1157

[FLINK-2720][storm-compatibility]Add Storm-CountMetric for storm metrics

I added the Storm-CountMetric for the first step about storm metrics.
1.Do a wrapper `FlinkCountMetric` for the CountMetric.
2.There is a real metric LongCounter in FlinkCountMetric class.
3.Push the RuntimeContext in `FlinkTopologyContext` for registering the 
metric.
4.Add a simple ut for the  `FlinkCountMetric` .

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/HuangWHWHW/flink FLINK-2720

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/flink/pull/1157.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #1157


commit 13c696beef318ef947b0e616fa9e29258bcbbf86
Author: HuangWHWHW <404823...@qq.com>
Date:   2015-09-21T13:22:02Z

[FLINK-2720][storm-compatibility]Add Storm-CountMetric for storm metrics




> Add Storm-CountMetric in flink-stormcompatibility
> -
>
> Key: FLINK-2720
> URL: https://issues.apache.org/jira/browse/FLINK-2720
> Project: Flink
>  Issue Type: New Feature
>  Components: Storm Compatibility
>Reporter: Huang Wei
>Assignee: Huang Wei
> Fix For: 0.10
>
>   Original Estimate: 168h
>  Remaining Estimate: 168h
>
> Add the CountMetric for the first step of storm metrics:
> 1.Do a wrapper FlinkCountMetric for CountMetric
> 2.push the RuntimeContext in FlinkTopologyContext to use `addAccumulator` 
> method for registering the metric.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)