flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (FLINK-9060) Deleting state using KeyedStateBackend.getKeys() throws Exception
Date Tue, 27 Mar 2018 11:29:00 GMT

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

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

Github user sihuazhou commented on the issue:

    https://github.com/apache/flink/pull/5751
  
    Hi @kl0u ! I have changed the code, could you please have a look when you have time?


> 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)

Mime
View raw message