You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Kostas Kloudas (JIRA)" <ji...@apache.org> on 2018/03/29 12:29:00 UTC
[jira] [Closed] (FLINK-9060) Deleting state using
KeyedStateBackend.getKeys() throws Exception
[ https://issues.apache.org/jira/browse/FLINK-9060?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Kostas Kloudas closed FLINK-9060.
---------------------------------
Resolution: Fixed
Merged on master with 62bbada10847609884f14ceee74588ab2c8f3d8c
and on release-1.5 with caac61833ddc51100703465555590f14dcf9db2b
> Deleting state using KeyedStateBackend.getKeys() throws Exception
> -----------------------------------------------------------------
>
> Key: FLINK-9060
> URL: https://issues.apache.org/jira/browse/FLINK-9060
> Project: Flink
> Issue Type: Bug
> Components: State Backends, Checkpointing
> Reporter: Aljoscha Krettek
> Assignee: Sihua Zhou
> Priority: Blocker
> Fix For: 1.5.0
>
>
> Adding this test to {{StateBackendTestBase}} showcases the problem:
> {code}
> @Test
> public void testConcurrentModificationWithGetKeys() throws Exception {
> AbstractKeyedStateBackend<Integer> backend = createKeyedBackend(IntSerializer.INSTANCE);
> try {
> ListStateDescriptor<String> listStateDescriptor =
> new ListStateDescriptor<>("foo", StringSerializer.INSTANCE);
> backend.setCurrentKey(1);
> backend
> .getPartitionedState(VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, listStateDescriptor)
> .add("Hello");
> backend.setCurrentKey(2);
> backend
> .getPartitionedState(VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, listStateDescriptor)
> .add("Ciao");
> Stream<Integer> keys = backend
> .getKeys(listStateDescriptor.getName(), VoidNamespace.INSTANCE);
> keys.forEach((key) -> {
> backend.setCurrentKey(key);
> try {
> backend
> .getPartitionedState(
> VoidNamespace.INSTANCE,
> VoidNamespaceSerializer.INSTANCE,
> listStateDescriptor)
> .clear();
> } catch (Exception e) {
> e.printStackTrace();
> }
> });
> }
> finally {
> IOUtils.closeQuietly(backend);
> backend.dispose();
> }
> }
> {code}
> This should work because one of the use cases of {{getKeys()}} and {{applyToAllKeys()}} is to do stuff for every key, which includes deleting them.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)