Errors in QueryableState sample code?

[Ken Krugler <kk...@transpac.com>]

Hi all,

I was looking at the Example <https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/state/queryable_state.html#example> section of Querying State <https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/state/queryable_state.html#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
+1 530-210-6378
http://www.scaleunlimited.com
Custom big data solutions & training
Flink, Solr, Hadoop, Cascading & Cassandra

Re: Errors in QueryableState sample code?

[vino yang <ya...@gmail.com>]

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 <kk...@transpac.com> 于2018年9月20日周四 上午6:27写道：

> Hi all,
>
> I was looking at the Example
> <https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/state/queryable_state.html#example> section
> of Querying State
> <https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/state/queryable_state.html#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 valueresultFuture.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
> +1 530-210-6378
> http://www.scaleunlimited.com
> Custom big data solutions & training
> Flink, Solr, Hadoop, Cascading & Cassandra
>
>