{"action":"created","issue":{"url":"https://api.github.com/repos/hazelcast/hazelcast/issues/10065","repository_url":"https://api.github.com/repos/hazelcast/hazelcast","labels_url":"https://api.github.com/repos/hazelcast/hazelcast/issues/10065/labels{/name}","comments_url":"https://api.github.com/repos/hazelcast/hazelcast/issues/10065/comments","events_url":"https://api.github.com/repos/hazelcast/hazelcast/issues/10065/events","html_url":"https://github.com/hazelcast/hazelcast/issues/10065","id":213711686,"node_id":"MDU6SXNzdWUyMTM3MTE2ODY=","number":10065,"title":"Multiple entry EntryProcessors do not respect locked entries","user":{"login":"mmedenjak","id":17194871,"node_id":"MDQ6VXNlcjE3MTk0ODcx","avatar_url":"https://avatars0.githubusercontent.com/u/17194871?v=4","gravatar_id":"","url":"https://api.github.com/users/mmedenjak","html_url":"https://github.com/mmedenjak","followers_url":"https://api.github.com/users/mmedenjak/followers","following_url":"https://api.github.com/users/mmedenjak/following{/other_user}","gists_url":"https://api.github.com/users/mmedenjak/gists{/gist_id}","starred_url":"https://api.github.com/users/mmedenjak/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mmedenjak/subscriptions","organizations_url":"https://api.github.com/users/mmedenjak/orgs","repos_url":"https://api.github.com/users/mmedenjak/repos","events_url":"https://api.github.com/users/mmedenjak/events{/privacy}","received_events_url":"https://api.github.com/users/mmedenjak/received_events","type":"User","site_admin":false},"labels":[{"id":795813601,"node_id":"MDU6TGFiZWw3OTU4MTM2MDE=","url":"https://api.github.com/repos/hazelcast/hazelcast/labels/Estimation:%20M","name":"Estimation: M","color":"fef2c0","default":false},{"id":838974591,"node_id":"MDU6TGFiZWw4Mzg5NzQ1OTE=","url":"https://api.github.com/repos/hazelcast/hazelcast/labels/Interface:%20IMap","name":"Interface: IMap","color":"efefef","default":false},{"id":789013630,"node_id":"MDU6TGFiZWw3ODkwMTM2MzA=","url":"https://api.github.com/repos/hazelcast/hazelcast/labels/Priority:%20Medium","name":"Priority: Medium","color":"e99695","default":false},{"id":137752575,"node_id":"MDU6TGFiZWwxMzc3NTI1NzU=","url":"https://api.github.com/repos/hazelcast/hazelcast/labels/Team:%20Core","name":"Team: Core","color":"109def","default":false},{"id":1740690,"node_id":"MDU6TGFiZWwxNzQwNjkw","url":"https://api.github.com/repos/hazelcast/hazelcast/labels/Type:%20Defect","name":"Type: Defect","color":"eb6420","default":false}],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":{"url":"https://api.github.com/repos/hazelcast/hazelcast/milestones/94","html_url":"https://github.com/hazelcast/hazelcast/milestone/94","labels_url":"https://api.github.com/repos/hazelcast/hazelcast/milestones/94/labels","id":2488368,"node_id":"MDk6TWlsZXN0b25lMjQ4ODM2OA==","number":94,"title":"4.1","description":null,"creator":{"login":"sancar","id":570462,"node_id":"MDQ6VXNlcjU3MDQ2Mg==","avatar_url":"https://avatars0.githubusercontent.com/u/570462?v=4","gravatar_id":"","url":"https://api.github.com/users/sancar","html_url":"https://github.com/sancar","followers_url":"https://api.github.com/users/sancar/followers","following_url":"https://api.github.com/users/sancar/following{/other_user}","gists_url":"https://api.github.com/users/sancar/gists{/gist_id}","starred_url":"https://api.github.com/users/sancar/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sancar/subscriptions","organizations_url":"https://api.github.com/users/sancar/orgs","repos_url":"https://api.github.com/users/sancar/repos","events_url":"https://api.github.com/users/sancar/events{/privacy}","received_events_url":"https://api.github.com/users/sancar/received_events","type":"User","site_admin":false},"open_issues":16,"closed_issues":2,"state":"open","created_at":"2017-04-28T14:45:30Z","updated_at":"2018-07-16T08:05:18Z","due_on":null,"closed_at":null},"comments":3,"created_at":"2017-03-13T09:25:06Z","updated_at":"2018-07-30T16:37:13Z","closed_at":null,"author_association":"CONTRIBUTOR","body":"In case of multiple entry `EntryProcessor` (`MultipleEntryOperation` and `PartitionWideEntryOperation`), the processor will execute on the entry regardless if it is locked or not.\r\n\r\nCurrently the entire EntryProcessor is run on the partition thread so in theory we could check if all affected entries are locked and park the operation if so. On unlock & notification we re-check and again park the operation until all affected entries are unlocked. This sounds simple but would require a special `WaitNotifyKey` for a partition wide-notification and it would require some changes in the operation system since it now expects that one notification can notify a single waiting key.\r\n\r\nWith the introduction of EP `Offloadable` the situation is more complex since the EP can now be interleaved with other operations (also lock and unlock). This is why we need to check if the entry is locked before running the EP logic on every key. If the entry is locked, there are two possible approaches:\r\n\r\n1. Park the EP operation. This requires substantial changes in the operation system since the operation isn't supposed to be parked/unparked after it started running.\r\n2. Create a EntryOperation for the locked key and continue processing the unlocked keys. Once the EntryOperation is finished update the value on the multiple-entry operation and check if all entries have been processed. This is more in line with existing operations where parts of the work are forked onto other operations and using callbacks we notify the parent operation (e.g. in WAN - `WanSyncOperation`).\r\n\r\nThese tests demonstrate the difference introduced by adding `Offloadable` interface for EP which relies on locking keys for entry update.\r\n\r\n```\r\n@RunWith(HazelcastParallelClassRunner.class)\r\n@Category({QuickTest.class, ParallelTest.class})\r\npublic class MixedEntryProcessorsTest extends HazelcastTestSupport {\r\n\r\n    public static final String MAP_NAME = \"EntryProcessorOffloadableTest\";\r\n\r\n    private HazelcastInstance[] instances;\r\n\r\n    @Override\r\n    public Config getConfig() {\r\n        Config config = super.getConfig();\r\n        MapConfig mapConfig = new MapConfig(MAP_NAME);\r\n        mapConfig.setInMemoryFormat(OBJECT);\r\n        mapConfig.setBackupCount(1);\r\n        config.addMapConfig(mapConfig);\r\n        return config;\r\n    }\r\n\r\n    @Before\r\n    public void before() {\r\n        TestHazelcastInstanceFactory factory = createHazelcastInstanceFactory(3);\r\n        instances = factory.newInstances(getConfig());\r\n    }\r\n\r\n    @Test\r\n    public void withOffloading() {\r\n        runTest(true);\r\n    }\r\n\r\n    @Test\r\n    public void withoutOffloading() {\r\n        runTest(false);\r\n    }\r\n\r\n    private void runTest(final boolean offload) {\r\n        final String key = generateKeyOwnedBy(instances[0]);\r\n        final int initialMoney = 100000;\r\n        User user = new User(initialMoney, 0, 0);\r\n\r\n        final IMap<Object, Object> map = instances[1].getMap(MAP_NAME);\r\n        map.put(key, user);\r\n        int boughtItems = 0;\r\n\r\n        for (int i = 0; i < 100; i++) {\r\n            map.executeOnEntries(new BuyItem());\r\n            boughtItems++;\r\n        }\r\n\r\n        final AtomicBoolean shouldRun = new AtomicBoolean(true);\r\n        spawn(new Runnable() {\r\n            @Override\r\n            public void run() {\r\n                map.executeOnKey(key, new CalculateBalance(offload));\r\n                shouldRun.set(false);\r\n            }\r\n        });\r\n\r\n        while (shouldRun.get()) {\r\n            map.executeOnEntries(new BuyItem());\r\n            boughtItems++;\r\n        }\r\n\r\n        map.executeOnKey(key, new CalculateBalance(offload));\r\n        final User o = (User) map.get(key);\r\n        System.out.println(\"CHECKING \" + o);\r\n        Assert.assertEquals(boughtItems, o.paidItems);\r\n    }\r\n\r\n\r\n    private static class User implements Serializable {\r\n\r\n        public int money;\r\n        public int itemsToPay;\r\n        public int paidItems;\r\n\r\n        User() {\r\n        }\r\n\r\n        public User(int money, int itemsToPay, int paidItems) {\r\n            this.money = money;\r\n            this.itemsToPay = itemsToPay;\r\n            this.paidItems = paidItems;\r\n        }\r\n\r\n        @Override\r\n        public boolean equals(Object o) {\r\n            if (this == o) {\r\n                return true;\r\n            }\r\n            if (o == null || getClass() != o.getClass()) {\r\n                return false;\r\n            }\r\n\r\n            User user = (User) o;\r\n\r\n            if (money != user.money) {\r\n                return false;\r\n            }\r\n            if (itemsToPay != user.itemsToPay) {\r\n                return false;\r\n            }\r\n            return paidItems == user.paidItems;\r\n        }\r\n\r\n        @Override\r\n        public int hashCode() {\r\n            int result = money;\r\n            result = 31 * result + itemsToPay;\r\n            result = 31 * result + paidItems;\r\n            return result;\r\n        }\r\n\r\n        @Override\r\n        public String toString() {\r\n            final StringBuffer sb = new StringBuffer(\"User{\");\r\n            sb.append(\"money=\").append(money);\r\n            sb.append(\", itemsToPay=\").append(itemsToPay);\r\n            sb.append(\", paidItems=\").append(paidItems);\r\n            sb.append('}');\r\n            return sb.toString();\r\n        }\r\n    }\r\n\r\n    private static class BuyItem implements EntryProcessor<String, User>, EntryBackupProcessor<String, User> {\r\n        @Override\r\n        public Object process(final Map.Entry<String, User> entry) {\r\n            final User value = entry.getValue();\r\n            value.itemsToPay++;\r\n            entry.setValue(value);\r\n            return value;\r\n        }\r\n\r\n        @Override\r\n        public EntryBackupProcessor<String, User> getBackupProcessor() {\r\n            return this;\r\n        }\r\n\r\n        @Override\r\n        public void processBackup(Map.Entry<String, User> entry) {\r\n            process(entry);\r\n        }\r\n    }\r\n\r\n    private static class CalculateBalance implements EntryProcessor<String, User>,\r\n            Offloadable,\r\n            EntryBackupProcessor<String, User> {\r\n        private boolean offload;\r\n\r\n        public CalculateBalance() {\r\n        }\r\n\r\n        public CalculateBalance(boolean offload) {\r\n            this.offload = offload;\r\n        }\r\n\r\n        @Override\r\n        public Object process(final Map.Entry<String, User> entry) {\r\n            // this is going to take a long time\r\n            try {\r\n                TimeUnit.SECONDS.sleep(1);\r\n            } catch (InterruptedException e) {\r\n                e.printStackTrace();\r\n            }\r\n            final User value = entry.getValue();\r\n            value.money -= value.itemsToPay;\r\n            value.paidItems += value.itemsToPay;\r\n            value.itemsToPay = 0;\r\n            entry.setValue(value);\r\n            return value;\r\n        }\r\n\r\n        @Override\r\n        public EntryBackupProcessor<String, User> getBackupProcessor() {\r\n            return this;\r\n        }\r\n\r\n        @Override\r\n        public String getExecutorName() {\r\n            return offload ? Offloadable.DEFAULT_ASYNC_EXECUTOR : Offloadable.NO_OFFLOADING;\r\n        }\r\n\r\n        @Override\r\n        public void processBackup(Map.Entry<String, User> entry) {\r\n            process(entry);\r\n        }\r\n    }\r\n\r\n}\r\n```"},"comment":{"url":"https://api.github.com/repos/hazelcast/hazelcast/issues/comments/408928264","html_url":"https://github.com/hazelcast/hazelcast/issues/10065#issuecomment-408928264","issue_url":"https://api.github.com/repos/hazelcast/hazelcast/issues/10065","id":408928264,"node_id":"MDEyOklzc3VlQ29tbWVudDQwODkyODI2NA==","user":{"login":"migralito","id":2132281,"node_id":"MDQ6VXNlcjIxMzIyODE=","avatar_url":"https://avatars0.githubusercontent.com/u/2132281?v=4","gravatar_id":"","url":"https://api.github.com/users/migralito","html_url":"https://github.com/migralito","followers_url":"https://api.github.com/users/migralito/followers","following_url":"https://api.github.com/users/migralito/following{/other_user}","gists_url":"https://api.github.com/users/migralito/gists{/gist_id}","starred_url":"https://api.github.com/users/migralito/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/migralito/subscriptions","organizations_url":"https://api.github.com/users/migralito/orgs","repos_url":"https://api.github.com/users/migralito/repos","events_url":"https://api.github.com/users/migralito/events{/privacy}","received_events_url":"https://api.github.com/users/migralito/received_events","type":"User","site_admin":false},"created_at":"2018-07-30T16:37:13Z","updated_at":"2018-07-30T16:37:13Z","author_association":"NONE","body":"I got REALLY confused about the fact that EntryProcessor javadoc states that entries passed to it will have the corresponding key locked. How is it possible that javadoc says something that it is not?!?!?!\r\n\r\n> ...At the time you receive it the entry is locked and not released until the EntryProcessor completes..."}}
{"actor":{"display_login":"migralito"}}