Errors in QueryableState sample code?

[Ken Krugler]

Hi all,

I was looking at the Example section of Querying State:

QueryableStateClient client = new QueryableStateClient(tmHostname, proxyPort);

// the state descriptor of the state to be fetched.
ValueStateDescriptor<Tuple2<Long, Long>> descriptor =
        new ValueStateDescriptor<>(
          "average",
          TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {}));

CompletableFuture<ValueState<Tuple2<Long, Long>>> resultFuture =
        client.getKvState(jobId, "query-name", key, BasicTypeInfo.LONG_TYPE_INFO, descriptor);

// now handle the returned value
resultFuture.thenAccept(response -> {
        try {
            Tuple2<Long, Long> res = response.get();
        } catch (Exception e) {
            e.printStackTrace();
        }
});

The two issues I ran into while trying out this code snippet were:

1. The query request would fail with a NPE, because the client had a null ExecutionConfig.

If I added:

            client.setExecutionConfig(new ExecutionConfig());

Then everything seemed OK, but this doesn’t feel like the right way to solve that problem :)

2. The call to response.get() returns a ValueState<Tuple2<Long, Long>>, not the Tuple2 itself.

So it seems like there’s a missing “.value()”.

Regards,

— Ken

--------------------------

Ken Krugler
[phone]
http://www.scaleunlimited.com
Custom big data solutions & training
Flink, Solr, Hadoop, Cascading & Cassandra

[vino yang]

Hi Ken, 

About the first question, the way of fixing is right.

About the second question, you are right, the "response.get()" missed 'ValueState<>'. 

Can you create a JIRA issue or fix it if you want.

Thanks, vino.

Ken Krugler <[hidden email]> 于2018年9月20日周四 上午6:27写道：

Hi all,

I was looking at the Example section of Querying State:

QueryableStateClient client = new QueryableStateClient(tmHostname, proxyPort);

// the state descriptor of the state to be fetched.
ValueStateDescriptor<Tuple2<Long, Long>> descriptor =
        new ValueStateDescriptor<>(
          "average",
          TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {}));

CompletableFuture<ValueState<Tuple2<Long, Long>>> resultFuture =
        client.getKvState(jobId, "query-name", key, BasicTypeInfo.LONG_TYPE_INFO, descriptor);

// now handle the returned value
resultFuture.thenAccept(response -> {
        try {
            Tuple2<Long, Long> res = response.get();
        } catch (Exception e) {
            e.printStackTrace();
        }
});

The two issues I ran into while trying out this code snippet were:

1. The query request would fail with a NPE, because the client had a null ExecutionConfig.

If I added:

            client.setExecutionConfig(new ExecutionConfig());

Then everything seemed OK, but this doesn’t feel like the right way to solve that problem :)

2. The call to response.get() returns a ValueState<Tuple2<Long, Long>>, not the Tuple2 itself.

So it seems like there’s a missing “.value()”.

Regards,

— Ken

--------------------------

Ken Krugler
[phone]
http://www.scaleunlimited.com
Custom big data solutions & training
Flink, Solr, Hadoop, Cascading & Cassandra

... [show rest of quote]