How to dynamically initialize flink metrics in invoke method and then reuse it？

[wangl...@geekplus.com.cn]

In one flink operator, i want to initialize multiple flink metrics according to 
message content. 
As the code below.

public void invoke(ObjectNode node, Context context) throws Exception {

    String tableName = node.get("metadata").get("topic").asText();
    Meter meter = getRuntimeContext().getMetricGroup().meter(tableName, new 
MeterView(10));
    meter.markEvent();
    log.info("### counter: " + meter.toString() + "\t" +  meter.getCount());


But in this way every invoke call will initialize a new metrics and the count 
will be from zero again.
How can i reuse the metric initialized before?

Thanks,
Lei


wangl...@geekplus.com.cn