alex-plekhanov commented on code in PR #10393:
URL: https://github.com/apache/ignite/pull/10393#discussion_r1105521754
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java:
##########
@@ -1399,7 +1399,7 @@ private int partition(CacheObjectContext ctx, @Nullable
GridCacheContext cctx, O
throws IgniteCheckedException {
switch (type) {
case BinaryObjectImpl.TYPE_BINARY:
- return new BinaryObjectImpl(binaryContext(), bytes, 0);
+ return new BinaryObjectImpl(binaryContext(), bytes, ctx);
Review Comment:
Do we really need it if we don't transform the keys?
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transform/AbstractCacheObjectsTransformationTest.java:
##########
@@ -0,0 +1,435 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.transform;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.binary.BinaryObject;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.events.CacheObjectTransformedEvent;
+import org.apache.ignite.events.EventType;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.spi.transform.CacheObjectTransformerAdapter;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+
+/**
+ * Leak test.
+ */
+public abstract class AbstractCacheObjectsTransformationTest extends
GridCommonAbstractTest {
+ /** Cache name. */
+ protected static final String CACHE_NAME = "data";
+
+ /** Nodes count. */
+ protected static final int NODES = 3;
+
+ /** Key. */
+ protected int key;
+
+ /** Event queue. */
+ protected final ConcurrentLinkedDeque<CacheObjectTransformedEvent>
evtQueue = new ConcurrentLinkedDeque<>();
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ stopAllGrids();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.setCacheConfiguration(cacheConfiguration());
+ cfg.setIncludeEventTypes(EventType.EVT_CACHE_OBJECT_TRANSFORMED);
+
+ return cfg;
+ }
+
+ /**
+ * Gets cache configuration.
+ *
+ * @return Data cache configuration.
+ */
+ protected CacheConfiguration cacheConfiguration() {
Review Comment:
Let's use generics (ar least `CacheConfiguration<?, ?>` can be used) here
and in overloaded method to avoid IDE warnings.
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transform/AbstractCacheObjectsTransformationTest.java:
##########
@@ -0,0 +1,435 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.transform;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.binary.BinaryObject;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.events.CacheObjectTransformedEvent;
+import org.apache.ignite.events.EventType;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.spi.transform.CacheObjectTransformerAdapter;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+
+/**
+ * Leak test.
+ */
+public abstract class AbstractCacheObjectsTransformationTest extends
GridCommonAbstractTest {
+ /** Cache name. */
+ protected static final String CACHE_NAME = "data";
+
+ /** Nodes count. */
+ protected static final int NODES = 3;
+
+ /** Key. */
+ protected int key;
+
+ /** Event queue. */
+ protected final ConcurrentLinkedDeque<CacheObjectTransformedEvent>
evtQueue = new ConcurrentLinkedDeque<>();
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ stopAllGrids();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.setCacheConfiguration(cacheConfiguration());
+ cfg.setIncludeEventTypes(EventType.EVT_CACHE_OBJECT_TRANSFORMED);
+
+ return cfg;
+ }
+
+ /**
+ * Gets cache configuration.
+ *
+ * @return Data cache configuration.
+ */
+ protected CacheConfiguration cacheConfiguration() {
+ CacheConfiguration cfg = defaultCacheConfiguration();
+
+ cfg.setName(CACHE_NAME);
+ cfg.setBackups(NODES);
+ cfg.setReadFromBackup(true);
+ cfg.setWriteSynchronizationMode(FULL_SYNC);
+ cfg.setAffinity(new RendezvousAffinityFunction(false, 1)); //
Simplifies event calculation.
+
+ return cfg;
+ }
+
+ /**
+ *
+ */
+ protected Ignite prepareCluster() throws Exception {
+ Ignite ignite = startGrids(NODES);
+
+ awaitPartitionMapExchange();
+
+ ignite.events().remoteListen(
+ (uuid, evt) -> {
+ assertTrue(evt instanceof CacheObjectTransformedEvent);
+
+ evtQueue.add((CacheObjectTransformedEvent)evt);
+
+ return true;
+ },
+ null,
+ EventType.EVT_CACHE_OBJECT_TRANSFORMED);
+
+ return ignite;
+ }
+
+ /**
+ *
+ */
+ protected void putAndCheck(Object val, boolean transformableKey, boolean
transformableVal, boolean reversed) {
+ boolean binarizableVal = !(val instanceof String || val instanceof
Integer || val instanceof Object[] ||
+ val instanceof int[] || val instanceof Collection);
+
+ boolean binarizableColVal = (val instanceof Object[] && !(val
instanceof String[] || val instanceof int[])) ||
+ (val instanceof Collection && !(
+ ((Iterable<?>)val).iterator().next() instanceof String ||
Review Comment:
`(Iterable<?>)val).iterator().next()` can be simplified to
`F.first((Iterable<?>)val)`
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transform/AbstractCacheObjectsTransformationTest.java:
##########
@@ -0,0 +1,435 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.transform;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.binary.BinaryObject;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.events.CacheObjectTransformedEvent;
+import org.apache.ignite.events.EventType;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.spi.transform.CacheObjectTransformerAdapter;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+
+/**
+ * Leak test.
+ */
+public abstract class AbstractCacheObjectsTransformationTest extends
GridCommonAbstractTest {
+ /** Cache name. */
+ protected static final String CACHE_NAME = "data";
+
+ /** Nodes count. */
+ protected static final int NODES = 3;
+
+ /** Key. */
+ protected int key;
+
+ /** Event queue. */
+ protected final ConcurrentLinkedDeque<CacheObjectTransformedEvent>
evtQueue = new ConcurrentLinkedDeque<>();
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ stopAllGrids();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.setCacheConfiguration(cacheConfiguration());
+ cfg.setIncludeEventTypes(EventType.EVT_CACHE_OBJECT_TRANSFORMED);
+
+ return cfg;
+ }
+
+ /**
+ * Gets cache configuration.
+ *
+ * @return Data cache configuration.
+ */
+ protected CacheConfiguration cacheConfiguration() {
+ CacheConfiguration cfg = defaultCacheConfiguration();
+
+ cfg.setName(CACHE_NAME);
+ cfg.setBackups(NODES);
+ cfg.setReadFromBackup(true);
+ cfg.setWriteSynchronizationMode(FULL_SYNC);
+ cfg.setAffinity(new RendezvousAffinityFunction(false, 1)); //
Simplifies event calculation.
+
+ return cfg;
+ }
+
+ /**
+ *
+ */
+ protected Ignite prepareCluster() throws Exception {
+ Ignite ignite = startGrids(NODES);
+
+ awaitPartitionMapExchange();
+
+ ignite.events().remoteListen(
+ (uuid, evt) -> {
+ assertTrue(evt instanceof CacheObjectTransformedEvent);
+
+ evtQueue.add((CacheObjectTransformedEvent)evt);
+
+ return true;
+ },
+ null,
+ EventType.EVT_CACHE_OBJECT_TRANSFORMED);
+
+ return ignite;
+ }
+
+ /**
+ *
+ */
+ protected void putAndCheck(Object val, boolean transformableKey, boolean
transformableVal, boolean reversed) {
+ boolean binarizableVal = !(val instanceof String || val instanceof
Integer || val instanceof Object[] ||
+ val instanceof int[] || val instanceof Collection);
+
+ boolean binarizableColVal = (val instanceof Object[] && !(val
instanceof String[] || val instanceof int[])) ||
+ (val instanceof Collection && !(
+ ((Iterable<?>)val).iterator().next() instanceof String ||
+ ((Iterable<?>)val).iterator().next() instanceof Integer)
+ );
+
+ boolean binaryVal = val instanceof BinaryObject;
+ boolean binaryColVal = val instanceof BinaryObject[] ||
+ (val instanceof Collection && ((Iterable<?>)val).iterator().next()
instanceof BinaryObject);
+
+ if (binaryVal)
+ assertTrue(binarizableVal);
+
+ if (binaryColVal)
+ assertTrue(binarizableColVal);
+
+ assertFalse(binaryVal && binaryColVal);
+ assertFalse(binarizableVal && binarizableColVal);
+
+ Ignite node = backupNode(0, CACHE_NAME); // Any key, besause of single
partition.
+
+ IgniteCache<Object, Object> cache = node.getOrCreateCache(CACHE_NAME);
+
+ Object k = reversed ? val : ++key;
+ Object v = reversed ? ++key : val;
+
+ cache.put(k, v);
+
+ checkPut(!reversed && binarizableVal, reversed ? transformableKey :
transformableVal);
+
+ checkGet(
+ k,
+ v,
+ !reversed && binaryVal,
+ !reversed && binaryColVal,
+ !reversed && binarizableVal,
+ !reversed && binarizableColVal,
+ reversed ? transformableKey : transformableVal);
+ }
+
+ /**
+ *
+ */
+ private void checkPut(
+ boolean binarizableVal,
+ boolean transformableVal) {
+ int transformed = transformableVal ? 1 : 0;
+ int transformCancelled = transformableVal ? 0 : 1;
+ int restored = transformableVal && binarizableVal ? NODES : 0; //
Binary array is required (e.g. to wait for proper Metadata)
+
+ checkEvents(transformed, transformCancelled, restored,
transformableVal);
+ }
+
+ /**
+ *
+ */
+ private void checkGet(
+ Object key,
+ Object expVal,
+ boolean binaryExpVal,
+ boolean binaryColExpVal,
+ boolean binarizableVal,
+ boolean binarizableColVal,
+ boolean transformableVal) {
+ for (Ignite node : G.allGrids()) {
+ for (boolean keepBinary : new boolean[] {true, false})
+ getWithCheck(
+ node,
+ key,
+ expVal,
+ binaryExpVal,
+ binaryColExpVal,
+ binarizableVal,
+ binarizableColVal,
+ transformableVal,
+ keepBinary);
+ }
+ }
+
+ /**
+ *
+ */
+ private void getWithCheck(
+ Ignite node,
+ Object key,
+ Object expVal,
+ boolean binaryExpVal,
+ boolean binaryColExpVal,
+ boolean binarizableVal,
+ boolean binarizableColVal,
+ boolean transformableVal,
+ boolean keepBinary) {
Review Comment:
`) {` should be on the new line
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transform/AbstractCacheObjectsTransformationTest.java:
##########
@@ -0,0 +1,435 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.transform;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.binary.BinaryObject;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.events.CacheObjectTransformedEvent;
+import org.apache.ignite.events.EventType;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.spi.transform.CacheObjectTransformerAdapter;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+
+/**
+ * Leak test.
+ */
+public abstract class AbstractCacheObjectsTransformationTest extends
GridCommonAbstractTest {
+ /** Cache name. */
+ protected static final String CACHE_NAME = "data";
+
+ /** Nodes count. */
+ protected static final int NODES = 3;
+
+ /** Key. */
+ protected int key;
+
+ /** Event queue. */
+ protected final ConcurrentLinkedDeque<CacheObjectTransformedEvent>
evtQueue = new ConcurrentLinkedDeque<>();
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ stopAllGrids();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.setCacheConfiguration(cacheConfiguration());
+ cfg.setIncludeEventTypes(EventType.EVT_CACHE_OBJECT_TRANSFORMED);
+
+ return cfg;
+ }
+
+ /**
+ * Gets cache configuration.
+ *
+ * @return Data cache configuration.
+ */
+ protected CacheConfiguration cacheConfiguration() {
+ CacheConfiguration cfg = defaultCacheConfiguration();
+
+ cfg.setName(CACHE_NAME);
+ cfg.setBackups(NODES);
+ cfg.setReadFromBackup(true);
+ cfg.setWriteSynchronizationMode(FULL_SYNC);
+ cfg.setAffinity(new RendezvousAffinityFunction(false, 1)); //
Simplifies event calculation.
+
+ return cfg;
+ }
+
+ /**
+ *
+ */
+ protected Ignite prepareCluster() throws Exception {
+ Ignite ignite = startGrids(NODES);
+
+ awaitPartitionMapExchange();
+
+ ignite.events().remoteListen(
+ (uuid, evt) -> {
+ assertTrue(evt instanceof CacheObjectTransformedEvent);
+
+ evtQueue.add((CacheObjectTransformedEvent)evt);
+
+ return true;
+ },
+ null,
+ EventType.EVT_CACHE_OBJECT_TRANSFORMED);
+
+ return ignite;
+ }
+
+ /**
+ *
+ */
+ protected void putAndCheck(Object val, boolean transformableKey, boolean
transformableVal, boolean reversed) {
+ boolean binarizableVal = !(val instanceof String || val instanceof
Integer || val instanceof Object[] ||
+ val instanceof int[] || val instanceof Collection);
+
+ boolean binarizableColVal = (val instanceof Object[] && !(val
instanceof String[] || val instanceof int[])) ||
+ (val instanceof Collection && !(
+ ((Iterable<?>)val).iterator().next() instanceof String ||
+ ((Iterable<?>)val).iterator().next() instanceof Integer)
+ );
+
+ boolean binaryVal = val instanceof BinaryObject;
+ boolean binaryColVal = val instanceof BinaryObject[] ||
+ (val instanceof Collection && ((Iterable<?>)val).iterator().next()
instanceof BinaryObject);
+
+ if (binaryVal)
+ assertTrue(binarizableVal);
+
+ if (binaryColVal)
+ assertTrue(binarizableColVal);
+
+ assertFalse(binaryVal && binaryColVal);
+ assertFalse(binarizableVal && binarizableColVal);
+
+ Ignite node = backupNode(0, CACHE_NAME); // Any key, besause of single
partition.
+
+ IgniteCache<Object, Object> cache = node.getOrCreateCache(CACHE_NAME);
+
+ Object k = reversed ? val : ++key;
+ Object v = reversed ? ++key : val;
+
+ cache.put(k, v);
+
+ checkPut(!reversed && binarizableVal, reversed ? transformableKey :
transformableVal);
+
+ checkGet(
+ k,
+ v,
+ !reversed && binaryVal,
+ !reversed && binaryColVal,
+ !reversed && binarizableVal,
+ !reversed && binarizableColVal,
+ reversed ? transformableKey : transformableVal);
+ }
+
+ /**
+ *
+ */
+ private void checkPut(
+ boolean binarizableVal,
+ boolean transformableVal) {
+ int transformed = transformableVal ? 1 : 0;
+ int transformCancelled = transformableVal ? 0 : 1;
+ int restored = transformableVal && binarizableVal ? NODES : 0; //
Binary array is required (e.g. to wait for proper Metadata)
+
+ checkEvents(transformed, transformCancelled, restored,
transformableVal);
+ }
+
+ /**
+ *
+ */
+ private void checkGet(
+ Object key,
+ Object expVal,
+ boolean binaryExpVal,
+ boolean binaryColExpVal,
+ boolean binarizableVal,
+ boolean binarizableColVal,
+ boolean transformableVal) {
Review Comment:
`) {` should be on the new line
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transform/AbstractCacheObjectsTransformationTest.java:
##########
@@ -0,0 +1,435 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.transform;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.binary.BinaryObject;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.events.CacheObjectTransformedEvent;
+import org.apache.ignite.events.EventType;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.spi.transform.CacheObjectTransformerAdapter;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+
+/**
+ * Leak test.
+ */
+public abstract class AbstractCacheObjectsTransformationTest extends
GridCommonAbstractTest {
+ /** Cache name. */
+ protected static final String CACHE_NAME = "data";
+
+ /** Nodes count. */
+ protected static final int NODES = 3;
+
+ /** Key. */
+ protected int key;
+
+ /** Event queue. */
+ protected final ConcurrentLinkedDeque<CacheObjectTransformedEvent>
evtQueue = new ConcurrentLinkedDeque<>();
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ stopAllGrids();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.setCacheConfiguration(cacheConfiguration());
+ cfg.setIncludeEventTypes(EventType.EVT_CACHE_OBJECT_TRANSFORMED);
+
+ return cfg;
+ }
+
+ /**
+ * Gets cache configuration.
+ *
+ * @return Data cache configuration.
+ */
+ protected CacheConfiguration cacheConfiguration() {
+ CacheConfiguration cfg = defaultCacheConfiguration();
+
+ cfg.setName(CACHE_NAME);
+ cfg.setBackups(NODES);
+ cfg.setReadFromBackup(true);
+ cfg.setWriteSynchronizationMode(FULL_SYNC);
+ cfg.setAffinity(new RendezvousAffinityFunction(false, 1)); //
Simplifies event calculation.
+
+ return cfg;
+ }
+
+ /**
+ *
+ */
+ protected Ignite prepareCluster() throws Exception {
+ Ignite ignite = startGrids(NODES);
+
+ awaitPartitionMapExchange();
+
+ ignite.events().remoteListen(
+ (uuid, evt) -> {
+ assertTrue(evt instanceof CacheObjectTransformedEvent);
+
+ evtQueue.add((CacheObjectTransformedEvent)evt);
+
+ return true;
+ },
+ null,
+ EventType.EVT_CACHE_OBJECT_TRANSFORMED);
+
+ return ignite;
+ }
+
+ /**
+ *
+ */
+ protected void putAndCheck(Object val, boolean transformableKey, boolean
transformableVal, boolean reversed) {
+ boolean binarizableVal = !(val instanceof String || val instanceof
Integer || val instanceof Object[] ||
+ val instanceof int[] || val instanceof Collection);
+
+ boolean binarizableColVal = (val instanceof Object[] && !(val
instanceof String[] || val instanceof int[])) ||
+ (val instanceof Collection && !(
+ ((Iterable<?>)val).iterator().next() instanceof String ||
+ ((Iterable<?>)val).iterator().next() instanceof Integer)
+ );
+
+ boolean binaryVal = val instanceof BinaryObject;
+ boolean binaryColVal = val instanceof BinaryObject[] ||
+ (val instanceof Collection && ((Iterable<?>)val).iterator().next()
instanceof BinaryObject);
+
+ if (binaryVal)
+ assertTrue(binarizableVal);
+
+ if (binaryColVal)
+ assertTrue(binarizableColVal);
+
+ assertFalse(binaryVal && binaryColVal);
+ assertFalse(binarizableVal && binarizableColVal);
+
+ Ignite node = backupNode(0, CACHE_NAME); // Any key, besause of single
partition.
+
+ IgniteCache<Object, Object> cache = node.getOrCreateCache(CACHE_NAME);
+
+ Object k = reversed ? val : ++key;
+ Object v = reversed ? ++key : val;
+
+ cache.put(k, v);
+
+ checkPut(!reversed && binarizableVal, reversed ? transformableKey :
transformableVal);
+
+ checkGet(
+ k,
+ v,
+ !reversed && binaryVal,
+ !reversed && binaryColVal,
+ !reversed && binarizableVal,
+ !reversed && binarizableColVal,
+ reversed ? transformableKey : transformableVal);
+ }
+
+ /**
+ *
+ */
+ private void checkPut(
+ boolean binarizableVal,
+ boolean transformableVal) {
Review Comment:
`) {` should be on the new line
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transform/CacheObjectsTransformationTest.java:
##########
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.transform;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import com.google.common.collect.Lists;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.binary.BinaryObject;
+import org.apache.ignite.binary.BinaryObjectBuilder;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+/**
+ *
+ */
+@RunWith(Parameterized.class)
+public class CacheObjectsTransformationTest extends
AbstractCacheObjectsTransformationTest {
+ /** Atomicity mode. */
+ @Parameterized.Parameter
+ public CacheAtomicityMode mode;
+
+ /** @return Test parameters. */
+ @Parameterized.Parameters(name = "mode={0}")
+ public static Collection<?> parameters() {
+ List<Object[]> res = new ArrayList<>();
+
+ for (CacheAtomicityMode mode : new CacheAtomicityMode[]
{CacheAtomicityMode.TRANSACTIONAL, CacheAtomicityMode.ATOMIC})
+ res.add(new Object[] {mode});
+
+ return res;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
+ return
super.getConfiguration(igniteInstanceName).setCacheObjectTransformer(new
ControllableCacheObjectTransformer());
+ }
+
+
+ /** {@inheritDoc} */
+ @Override protected CacheConfiguration cacheConfiguration() {
+ CacheConfiguration cfg = super.cacheConfiguration();
+
+ cfg.setAtomicityMode(mode);
+
+ return cfg;
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testTransformable() throws Exception {
+ ControllableCacheObjectTransformer.transformationShift(42);
+
+ assertFalse(ControllableCacheObjectTransformer.failOnTransformation());
+
+ doTest();
+
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testUntransformable() throws Exception {
+ ControllableCacheObjectTransformer.transformationShift(0); // Fail on
transformation.
+
+ assertTrue(ControllableCacheObjectTransformer.failOnTransformation());
+
+ doTest();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ private void doTest() throws Exception {
+ Ignite ignite = prepareCluster();
+
+ int i = -42; // Avoiding intersection with an incremental key.
+ int[] is = new int[] {i, i, i};
+ List<Integer> iList = Lists.newArrayList(i, i, i);
Review Comment:
`F.asList` instead of guava?
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transform/CacheObjectsTransformationTest.java:
##########
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.transform;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import com.google.common.collect.Lists;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.binary.BinaryObject;
+import org.apache.ignite.binary.BinaryObjectBuilder;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+/**
+ *
+ */
+@RunWith(Parameterized.class)
+public class CacheObjectsTransformationTest extends
AbstractCacheObjectsTransformationTest {
+ /** Atomicity mode. */
+ @Parameterized.Parameter
+ public CacheAtomicityMode mode;
+
+ /** @return Test parameters. */
+ @Parameterized.Parameters(name = "mode={0}")
+ public static Collection<?> parameters() {
+ List<Object[]> res = new ArrayList<>();
+
+ for (CacheAtomicityMode mode : new CacheAtomicityMode[]
{CacheAtomicityMode.TRANSACTIONAL, CacheAtomicityMode.ATOMIC})
+ res.add(new Object[] {mode});
+
+ return res;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
+ return
super.getConfiguration(igniteInstanceName).setCacheObjectTransformer(new
ControllableCacheObjectTransformer());
+ }
+
+
+ /** {@inheritDoc} */
+ @Override protected CacheConfiguration cacheConfiguration() {
+ CacheConfiguration cfg = super.cacheConfiguration();
+
+ cfg.setAtomicityMode(mode);
+
+ return cfg;
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testTransformable() throws Exception {
+ ControllableCacheObjectTransformer.transformationShift(42);
+
+ assertFalse(ControllableCacheObjectTransformer.failOnTransformation());
+
+ doTest();
+
Review Comment:
Redundant NL
##########
modules/compress/src/test/java/org/apache/ignite/internal/processors/cache/transform/CacheObjectsCompressionTest.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.transform;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.binary.BinaryObjectBuilder;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static
org.apache.ignite.internal.processors.cache.transform.AbstractCacheObjectsCompressionTest.CompressionTransformer.CompressionType;
+
+/**
+ *
+ */
+@RunWith(Parameterized.class)
+public class CacheObjectsCompressionTest extends
AbstractCacheObjectsCompressionTest {
+ /** Thin client. */
+ @Parameterized.Parameter
+ public CompressionType type;
+
+ /** @return Test parameters. */
+ @Parameterized.Parameters(name = "type={0}")
+ public static Collection<?> parameters() {
+ List<Object[]> res = new ArrayList<>();
+
+ for (CompressionType type : CompressionType.values())
+ res.add(new Object[] {type});
+
+ return res;
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testCompression() throws Exception {
+ try {
+ CompressionTransformer.type = type;
+
+ Ignite ignite = prepareCluster();
+
+ int i = 42;
+
+ putAndCheck(i, false); // No chances to compress integer.
+
+ String str = "Test string";
+
+ putAndCheck(str, false); // Too short string.
+
+ StringData sd = new StringData(str);
+
+ putAndCheck(sd, false); // Too short wrapped string.
+
+ List<Object> sdList = Collections.singletonList(sd);
+
+ putAndCheck(sdList, false); // Too short wrapped string.
+
+ StringBuilder sb = new StringBuilder();
+
+ for (int k = 0; k < 100; k++)
+ sb.append("AAAAAAAAAA");
+
+ String str2 = sb.toString();
+
+ putAndCheck(str2, type != CompressionType.DISABLED);
+
+ List<Object> list = new ArrayList<>();
+
+ list.add(new BinarizableData(str, null, i));
+
+ putAndCheck(list, false); // Too short list.
+
+ // Adding more elements.
+ list.add(new BinarizableData(str, null, i));
+ list.add(new BinarizableData(str, null, i));
+
+ putAndCheck(list, type != CompressionType.DISABLED); // Enough to
be compressed.
Review Comment:
As far as I know we can treat `str` as `handle` during binary object
marshalling (we don't do it for strings now, but format supports it and it can
be done without compatibility issues), in this case test can fail. At least
lets create a new string (`new String(str)` instead of `str`) or append `str`
with some char to ensure it will not be marshalled as `handle` if our
implementation of binary object marshaller will change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
