alex-plekhanov commented on code in PR #12128:
URL: https://github.com/apache/ignite/pull/12128#discussion_r2155040901
##########
modules/core/src/main/java/org/apache/ignite/internal/client/thin/TcpClientTransactions.java:
##########
@@ -280,5 +293,15 @@ ClientChannel clientChannel() {
boolean isClosed() {
return closed;
}
+
+ /** */
+ public TransactionConcurrency getConcurrency() {
+ return concurrency;
+ }
+
+ /** */
+ public TransactionIsolation getIsolation() {
Review Comment:
Use Ignite style getters name (without get)
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/cache/ImmutableArrayMap.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.platform.client.cache;
+
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Lightweight, array-backed read-only implementation of the {@link Map}
interface.
+ * <p>
+ * This class provides a fixed mapping of keys to values, based on the arrays
provided
+ * during construction. It does not support most modification operations such
as
+ * {@code put}, {@code remove}, or {@code clear}. Only {@code size}, {@code
isEmpty},
+ * {@code keySet}, and {@code values} are supported.
+ * <p>
+ * Intended for high-performance, read-only use cases where data is managed
externally.
+ *
+ * @param <K> Type of keys maintained by this map.
+ * @param <V> Type of mapped values.
+ */
+public class ImmutableArrayMap<K, V> implements Map<K, V>, Serializable {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** Backing immutable array for keys. */
+ private final Set<K> keys;
+
+ /** Backing immutable array for values. */
+ private final List<V> values;
+
+ /**
+ * Constructs a new {@code ImmutableArrayMap} with the given keys and
values.
+ * The arrays must be of the same length, and represent a one-to-one
mapping.
+ *
+ * @param keysArr Array of keys.
+ * @param valuesArr Array of values.
+ * @throws AssertionError if the array lengths differ.
+ */
+ public ImmutableArrayMap(K[] keysArr, V[] valuesArr) {
+ assert keysArr.length == valuesArr.length : "Arrays should be equal in
size!";
+
+ this.keys = new ImmutableArraySet<>(keysArr);
+ this.values = List.of(valuesArr);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int size() {
+ return keys.size();
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean isEmpty() {
+ return keys.isEmpty();
+ }
+
+ /**
+ * Unsupported operation.
+ *
+ * @throws UnsupportedOperationException always.
+ */
+ @Override public boolean containsKey(Object key) {
+ throw new UnsupportedOperationException("'containsKey' operation is
not supported by ImmutableArrayMap.");
+ }
+
+ /**
+ * Unsupported operation.
+ *
+ * @throws UnsupportedOperationException always.
+ */
+ @Override public boolean containsValue(Object val) {
+ throw new UnsupportedOperationException("'containsValue' operation is
not supported by ImmutableArrayMap.");
+ }
+
+ /**
+ * Unsupported operation.
+ *
+ * @throws UnsupportedOperationException always.
+ */
+ @Override public V get(Object key) {
+ throw new UnsupportedOperationException("'get' operation is not
supported by ImmutableArrayMap.");
+ }
+
+ /**
+ * Unsupported operation.
+ *
+ * @throws UnsupportedOperationException always.
+ */
+ @Override public V put(K key, V val) {
+ throw new UnsupportedOperationException("'put' operation is not
supported by ImmutableArrayMap.");
+ }
+
+ /**
+ * Unsupported operation.
+ *
+ * @throws UnsupportedOperationException always.
+ */
+ @Override public V remove(Object key) {
+ throw new UnsupportedOperationException("'remove' operation is not
supported by ImmutableArrayMap.");
+ }
+
+ /**
+ * Unsupported operation.
+ *
+ * @throws UnsupportedOperationException always.
+ */
+ @Override public void putAll(@NotNull Map<? extends K, ? extends V> m) {
+ throw new UnsupportedOperationException("'remove' operation is not
supported by ImmutableArrayMap.");
+ }
+
+ /**
+ * Unsupported operation.
+ *
+ * @throws UnsupportedOperationException always.
+ */
+ @Override public void clear() {
+ throw new UnsupportedOperationException("'clear' operation is not
supported by ImmutableArrayMap.");
+ }
+
+ /**
+ * Returns an immutable set view of the keys in this map.
+ *
+ * @return An immutable {@code Set} of keys.
+ */
+ @Override public @NotNull Set<K> keySet() {
+ return keys;
+ }
+
+ /**
+ * Returns an immutable collection view of the values in this map.
+ *
+ * @return An immutable {@code Collection} of values.
+ */
+ @Override public @NotNull Collection<V> values() {
+ return values;
+ }
+
+ /**
+ * Unsupported operation.
+ *
+ * @throws UnsupportedOperationException always.
+ */
+ @Override public @NotNull Set<Entry<K, V>> entrySet() {
+ throw new UnsupportedOperationException("'remove' operation is not
supported by ImmutableArrayMap.");
Review Comment:
'remove' -> 'entrySet'
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java:
##########
@@ -725,23 +759,22 @@ private <K, V> IgniteInternalFuture putAllAsync0(
if (opCtx != null && opCtx.hasDataCenterId()) {
assert drMap == null : drMap;
- assert map != null || invokeMap != null;
+ assert keySet != null || itInvokeVals != null;
dataCenterId = opCtx.dataCenterId();
}
else
dataCenterId = null;
- final Map<?, EntryProcessor<K, V, Object>> invokeMap0 = (Map<K,
EntryProcessor<K, V, Object>>)invokeMap;
-
if (log.isDebugEnabled())
- log.debug("Called putAllAsync(...) [tx=" + this + ", map=" + map +
", retval=" + retval + "]");
+ log.debug("Called putAllAsync(...) [tx=" + this + ", map=[" +
S.toString(Set.class, keySet) + ", " +
Review Comment:
`S.toString(Set.class, keySet)` looks redundant, just `keySet` is enough
##########
modules/core/src/test/java/org/apache/ignite/client/ClientOrderedCollectionWarnTest.java:
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.client;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import javax.cache.processor.EntryProcessor;
+import javax.cache.processor.MutableEntry;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.configuration.ClientConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.testframework.ListeningTestLogger;
+import org.apache.ignite.testframework.LogListener;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+import org.junit.Test;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
+import static org.apache.ignite.transactions.TransactionConcurrency.OPTIMISTIC;
+import static
org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static
org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
+import static org.apache.ignite.transactions.TransactionIsolation.SERIALIZABLE;
+
+/** */
+public class ClientOrderedCollectionWarnTest extends GridCommonAbstractTest {
+ /** */
+ private static final String WARN_LSNR_MSG = "Unordered %s java.util."; //
+
+ /** */
+ private static final LogListener MAP_WARN_LSNR =
+ LogListener.matches(String.format(WARN_LSNR_MSG,
"map")).times(1).build();
+
+ /** */
+ private static final LogListener SET_WARN_LSNR =
+ LogListener.matches(String.format(WARN_LSNR_MSG,
"collection")).times(1).build();
+
+ /** */
+ private final ListeningTestLogger testLog = new ListeningTestLogger(log);
+
+ /** */
+ private static IgniteEx ign;
+
+ /** */
+ private static IgniteClient cli;
+
+ /** */
+ private static ClientCache<Long, Long> cache;
+
+ /** */
+ @Override protected IgniteConfiguration getConfiguration(String
instanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(instanceName);
+
+ testLog.registerListener(MAP_WARN_LSNR);
+ testLog.registerListener(SET_WARN_LSNR);
+
+ cfg.setGridLogger(testLog);
+
+ return cfg;
+ }
+
+ /** */
+ private ClientConfiguration getClientConfiguration() {
+ return new
ClientConfiguration().setAddresses(Config.SERVER).setLogger(testLog);
+ }
+
+ /** */
+ private ClientCacheConfiguration getClientClientConfiguration() {
+ return new ClientCacheConfiguration()
+ .setName(DEFAULT_CACHE_NAME)
+ .setAtomicityMode(TRANSACTIONAL);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ ign = startGrid();
+ cli = Ignition.startClient(getClientConfiguration());
+
+ cache = cli.getOrCreateCache(getClientClientConfiguration());
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ super.afterTestsStopped();
+
+ ign.close();
+ cli.close();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ MAP_WARN_LSNR.reset();
+ SET_WARN_LSNR.reset();
+ }
+
+ /** */
+ @Test
+ public void putAll() throws Exception {
Review Comment:
Use identical tests naming. If `putAll` than `removeAll`, if `testRemoveAll`
than `testPutAll`.
##########
modules/core/src/main/java/org/apache/ignite/internal/client/thin/TcpClientCache.java:
##########
@@ -1616,4 +1650,77 @@ private void
checkDataReplicationSupported(ProtocolContext protocolCtx)
if
(!protocolCtx.isFeatureSupported(ProtocolBitmaskFeature.DATA_REPLICATION_OPERATIONS))
throw new
ClientFeatureNotSupportedByServerException(ProtocolBitmaskFeature.DATA_REPLICATION_OPERATIONS);
}
+
+ /**
+ * Warns if an unordered map is used in an operation that may lead to a
distributed deadlock
+ * during an explicit transaction.
+ * <p>
+ * This check is relevant only for explicit user-managed transactions.
Implicit transactions
+ * (such as those started automatically by the system) are not inspected
by this method.
+ * </p>
+ *
+ * @param m The map being used in the cache operation.
+ */
+ protected void warnIfUnordered(Map<?, ?> m) {
+ if (m == null || m.size() <= 1)
+ return;
+
+ TcpClientTransaction tx = transactions.tx();
+
+ // Only explicit transactions are checked
Review Comment:
Point at the end of line.
##########
modules/core/src/test/java/org/apache/ignite/client/ClientOrderedCollectionWarnTest.java:
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.client;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import javax.cache.processor.EntryProcessor;
+import javax.cache.processor.MutableEntry;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.configuration.ClientConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.testframework.ListeningTestLogger;
+import org.apache.ignite.testframework.LogListener;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+import org.junit.Test;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
+import static org.apache.ignite.transactions.TransactionConcurrency.OPTIMISTIC;
+import static
org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static
org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
+import static org.apache.ignite.transactions.TransactionIsolation.SERIALIZABLE;
+
+/** */
+public class ClientOrderedCollectionWarnTest extends GridCommonAbstractTest {
+ /** */
+ private static final String WARN_LSNR_MSG = "Unordered %s java.util."; //
+
+ /** */
+ private static final LogListener MAP_WARN_LSNR =
+ LogListener.matches(String.format(WARN_LSNR_MSG,
"map")).times(1).build();
+
+ /** */
+ private static final LogListener SET_WARN_LSNR =
+ LogListener.matches(String.format(WARN_LSNR_MSG,
"collection")).times(1).build();
+
+ /** */
+ private final ListeningTestLogger testLog = new ListeningTestLogger(log);
+
+ /** */
+ private static IgniteEx ign;
+
+ /** */
+ private static IgniteClient cli;
+
+ /** */
+ private static ClientCache<Long, Long> cache;
+
+ /** */
+ @Override protected IgniteConfiguration getConfiguration(String
instanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(instanceName);
+
+ testLog.registerListener(MAP_WARN_LSNR);
+ testLog.registerListener(SET_WARN_LSNR);
+
+ cfg.setGridLogger(testLog);
+
+ return cfg;
+ }
+
+ /** */
+ private ClientConfiguration getClientConfiguration() {
+ return new
ClientConfiguration().setAddresses(Config.SERVER).setLogger(testLog);
+ }
+
+ /** */
+ private ClientCacheConfiguration getClientClientConfiguration() {
+ return new ClientCacheConfiguration()
+ .setName(DEFAULT_CACHE_NAME)
+ .setAtomicityMode(TRANSACTIONAL);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ ign = startGrid();
+ cli = Ignition.startClient(getClientConfiguration());
+
+ cache = cli.getOrCreateCache(getClientClientConfiguration());
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ super.afterTestsStopped();
+
+ ign.close();
+ cli.close();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ MAP_WARN_LSNR.reset();
+ SET_WARN_LSNR.reset();
+ }
+
+ /** */
+ @Test
+ public void putAll() throws Exception {
+ Runnable cacheOpTreeMap = () -> cache.putAll(fillMap(new TreeMap<>()));
+ Runnable cacheOpHashMap = () -> cache.putAll(fillMap(new HashMap<>()));
+
+ testOp(cacheOpTreeMap, cacheOpHashMap, MAP_WARN_LSNR);
+ }
+
+ /** */
+ @Test
+ public void invokeAll() throws Exception {
+ Runnable cacheOpTreeSet = () -> cache.invokeAll(fillSet(new
TreeSet<>()), new TestEntryProcessor());
+ Runnable cacheOpHashSet = () -> cache.invokeAll(fillSet(new
HashSet<>()), new TestEntryProcessor());
+
+ testOp(cacheOpTreeSet, cacheOpHashSet, SET_WARN_LSNR);
+ }
+
+ /** */
+ @Test
+ public void testRemoveAll() throws Exception {
+ Runnable cacheOpTreeSet = () -> cache.removeAll(fillSet(new
TreeSet<>()));
+ Runnable cacheOpHashSet = () -> cache.removeAll(fillSet(new
HashSet<>()));
+
+ testOp(cacheOpTreeSet, cacheOpHashSet, SET_WARN_LSNR);
+ }
+
+ /** */
+ @Test
+ public void testGetAll() throws Exception {
+ Runnable cacheOpTreeSet = () -> cache.removeAll(fillSet(new
TreeSet<>()));
+ Runnable cacheOpHashSet = () -> cache.removeAll(fillSet(new
HashSet<>()));
+
+ testOp(cacheOpTreeSet, cacheOpHashSet, SET_WARN_LSNR);
+
+ SET_WARN_LSNR.reset();
+
+ withTx(cacheOpHashSet, PESSIMISTIC, READ_COMMITTED);
+
+ checkOp(false, SET_WARN_LSNR);
+ }
+
+ /** */
+ private void testOp(Runnable cacheOpWithOrdered, Runnable
cacheOpWithNoOrdered, LogListener lsnr) throws Exception {
+ cacheOpWithOrdered.run();
+
+ withTx(cacheOpWithOrdered, PESSIMISTIC, SERIALIZABLE);
+
+ withTx(cacheOpWithNoOrdered, OPTIMISTIC, SERIALIZABLE);
+
+ checkOp(false, lsnr);
+
+ withTx(cacheOpWithNoOrdered, PESSIMISTIC, READ_COMMITTED);
Review Comment:
Checked ordered and unordered collections with different tx
concurrency/isolation. I think it worth to add checks to show different
behavoiur with the same tx concurrency/isolation for ordered/unordered
collections.
##########
modules/core/src/test/java/org/apache/ignite/client/ClientOrderedCollectionWarnTest.java:
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.client;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import javax.cache.processor.EntryProcessor;
+import javax.cache.processor.MutableEntry;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.configuration.ClientConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.testframework.ListeningTestLogger;
+import org.apache.ignite.testframework.LogListener;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+import org.junit.Test;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
+import static org.apache.ignite.transactions.TransactionConcurrency.OPTIMISTIC;
+import static
org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static
org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
+import static org.apache.ignite.transactions.TransactionIsolation.SERIALIZABLE;
+
+/** */
+public class ClientOrderedCollectionWarnTest extends GridCommonAbstractTest {
+ /** */
+ private static final String WARN_LSNR_MSG = "Unordered %s java.util."; //
+
+ /** */
+ private static final LogListener MAP_WARN_LSNR =
+ LogListener.matches(String.format(WARN_LSNR_MSG,
"map")).times(1).build();
+
+ /** */
+ private static final LogListener SET_WARN_LSNR =
+ LogListener.matches(String.format(WARN_LSNR_MSG,
"collection")).times(1).build();
+
+ /** */
+ private final ListeningTestLogger testLog = new ListeningTestLogger(log);
+
+ /** */
+ private static IgniteEx ign;
+
+ /** */
+ private static IgniteClient cli;
+
+ /** */
+ private static ClientCache<Long, Long> cache;
+
+ /** */
+ @Override protected IgniteConfiguration getConfiguration(String
instanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(instanceName);
+
+ testLog.registerListener(MAP_WARN_LSNR);
+ testLog.registerListener(SET_WARN_LSNR);
+
+ cfg.setGridLogger(testLog);
Review Comment:
There must be different loggers for server and client. We should check that
related message is not present in server log (it's a main goal of current
patch).
##########
modules/core/src/test/java/org/apache/ignite/client/ClientOrderedCollectionWarnTest.java:
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.client;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import javax.cache.processor.EntryProcessor;
+import javax.cache.processor.MutableEntry;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.configuration.ClientConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.testframework.ListeningTestLogger;
+import org.apache.ignite.testframework.LogListener;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+import org.junit.Test;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
+import static org.apache.ignite.transactions.TransactionConcurrency.OPTIMISTIC;
+import static
org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static
org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
+import static org.apache.ignite.transactions.TransactionIsolation.SERIALIZABLE;
+
+/** */
+public class ClientOrderedCollectionWarnTest extends GridCommonAbstractTest {
+ /** */
+ private static final String WARN_LSNR_MSG = "Unordered %s java.util."; //
+
+ /** */
+ private static final LogListener MAP_WARN_LSNR =
+ LogListener.matches(String.format(WARN_LSNR_MSG,
"map")).times(1).build();
+
+ /** */
+ private static final LogListener SET_WARN_LSNR =
+ LogListener.matches(String.format(WARN_LSNR_MSG,
"collection")).times(1).build();
+
+ /** */
+ private final ListeningTestLogger testLog = new ListeningTestLogger(log);
+
+ /** */
+ private static IgniteEx ign;
+
+ /** */
+ private static IgniteClient cli;
+
+ /** */
+ private static ClientCache<Long, Long> cache;
+
+ /** */
+ @Override protected IgniteConfiguration getConfiguration(String
instanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(instanceName);
+
+ testLog.registerListener(MAP_WARN_LSNR);
+ testLog.registerListener(SET_WARN_LSNR);
+
+ cfg.setGridLogger(testLog);
+
+ return cfg;
+ }
+
+ /** */
+ private ClientConfiguration getClientConfiguration() {
+ return new
ClientConfiguration().setAddresses(Config.SERVER).setLogger(testLog);
+ }
+
+ /** */
+ private ClientCacheConfiguration getClientClientConfiguration() {
+ return new ClientCacheConfiguration()
+ .setName(DEFAULT_CACHE_NAME)
+ .setAtomicityMode(TRANSACTIONAL);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ ign = startGrid();
+ cli = Ignition.startClient(getClientConfiguration());
+
+ cache = cli.getOrCreateCache(getClientClientConfiguration());
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ super.afterTestsStopped();
+
+ ign.close();
+ cli.close();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ MAP_WARN_LSNR.reset();
+ SET_WARN_LSNR.reset();
+ }
+
+ /** */
+ @Test
+ public void putAll() throws Exception {
+ Runnable cacheOpTreeMap = () -> cache.putAll(fillMap(new TreeMap<>()));
+ Runnable cacheOpHashMap = () -> cache.putAll(fillMap(new HashMap<>()));
+
+ testOp(cacheOpTreeMap, cacheOpHashMap, MAP_WARN_LSNR);
+ }
+
+ /** */
+ @Test
+ public void invokeAll() throws Exception {
+ Runnable cacheOpTreeSet = () -> cache.invokeAll(fillSet(new
TreeSet<>()), new TestEntryProcessor());
+ Runnable cacheOpHashSet = () -> cache.invokeAll(fillSet(new
HashSet<>()), new TestEntryProcessor());
+
+ testOp(cacheOpTreeSet, cacheOpHashSet, SET_WARN_LSNR);
+ }
+
+ /** */
+ @Test
+ public void testRemoveAll() throws Exception {
+ Runnable cacheOpTreeSet = () -> cache.removeAll(fillSet(new
TreeSet<>()));
+ Runnable cacheOpHashSet = () -> cache.removeAll(fillSet(new
HashSet<>()));
+
+ testOp(cacheOpTreeSet, cacheOpHashSet, SET_WARN_LSNR);
+ }
+
+ /** */
+ @Test
+ public void testGetAll() throws Exception {
+ Runnable cacheOpTreeSet = () -> cache.removeAll(fillSet(new
TreeSet<>()));
Review Comment:
`testGetAll` but `removeAll` inside.
##########
modules/core/src/main/java/org/apache/ignite/internal/client/thin/TcpClientTransactions.java:
##########
@@ -280,5 +293,15 @@ ClientChannel clientChannel() {
boolean isClosed() {
return closed;
}
+
+ /** */
+ public TransactionConcurrency getConcurrency() {
Review Comment:
Use Ignite style getters name (without get)
##########
modules/core/src/test/java/org/apache/ignite/client/ClientOrderedCollectionWarnTest.java:
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.client;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import javax.cache.processor.EntryProcessor;
+import javax.cache.processor.MutableEntry;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.configuration.ClientConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.testframework.ListeningTestLogger;
+import org.apache.ignite.testframework.LogListener;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+import org.junit.Test;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
+import static org.apache.ignite.transactions.TransactionConcurrency.OPTIMISTIC;
+import static
org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static
org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
+import static org.apache.ignite.transactions.TransactionIsolation.SERIALIZABLE;
+
+/** */
+public class ClientOrderedCollectionWarnTest extends GridCommonAbstractTest {
+ /** */
+ private static final String WARN_LSNR_MSG = "Unordered %s java.util."; //
+
+ /** */
+ private static final LogListener MAP_WARN_LSNR =
+ LogListener.matches(String.format(WARN_LSNR_MSG,
"map")).times(1).build();
+
+ /** */
+ private static final LogListener SET_WARN_LSNR =
+ LogListener.matches(String.format(WARN_LSNR_MSG,
"collection")).times(1).build();
+
+ /** */
+ private final ListeningTestLogger testLog = new ListeningTestLogger(log);
+
+ /** */
+ private static IgniteEx ign;
+
+ /** */
+ private static IgniteClient cli;
+
+ /** */
+ private static ClientCache<Long, Long> cache;
+
+ /** */
+ @Override protected IgniteConfiguration getConfiguration(String
instanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(instanceName);
+
+ testLog.registerListener(MAP_WARN_LSNR);
+ testLog.registerListener(SET_WARN_LSNR);
+
+ cfg.setGridLogger(testLog);
+
+ return cfg;
+ }
+
+ /** */
+ private ClientConfiguration getClientConfiguration() {
+ return new
ClientConfiguration().setAddresses(Config.SERVER).setLogger(testLog);
+ }
+
+ /** */
+ private ClientCacheConfiguration getClientClientConfiguration() {
+ return new ClientCacheConfiguration()
+ .setName(DEFAULT_CACHE_NAME)
+ .setAtomicityMode(TRANSACTIONAL);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ ign = startGrid();
+ cli = Ignition.startClient(getClientConfiguration());
+
+ cache = cli.getOrCreateCache(getClientClientConfiguration());
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ super.afterTestsStopped();
+
+ ign.close();
+ cli.close();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ MAP_WARN_LSNR.reset();
+ SET_WARN_LSNR.reset();
+ }
+
+ /** */
+ @Test
+ public void putAll() throws Exception {
+ Runnable cacheOpTreeMap = () -> cache.putAll(fillMap(new TreeMap<>()));
+ Runnable cacheOpHashMap = () -> cache.putAll(fillMap(new HashMap<>()));
+
+ testOp(cacheOpTreeMap, cacheOpHashMap, MAP_WARN_LSNR);
+ }
+
+ /** */
+ @Test
+ public void invokeAll() throws Exception {
+ Runnable cacheOpTreeSet = () -> cache.invokeAll(fillSet(new
TreeSet<>()), new TestEntryProcessor());
+ Runnable cacheOpHashSet = () -> cache.invokeAll(fillSet(new
HashSet<>()), new TestEntryProcessor());
+
+ testOp(cacheOpTreeSet, cacheOpHashSet, SET_WARN_LSNR);
+ }
+
+ /** */
+ @Test
+ public void testRemoveAll() throws Exception {
+ Runnable cacheOpTreeSet = () -> cache.removeAll(fillSet(new
TreeSet<>()));
+ Runnable cacheOpHashSet = () -> cache.removeAll(fillSet(new
HashSet<>()));
+
+ testOp(cacheOpTreeSet, cacheOpHashSet, SET_WARN_LSNR);
+ }
+
+ /** */
+ @Test
+ public void testGetAll() throws Exception {
+ Runnable cacheOpTreeSet = () -> cache.removeAll(fillSet(new
TreeSet<>()));
+ Runnable cacheOpHashSet = () -> cache.removeAll(fillSet(new
HashSet<>()));
+
+ testOp(cacheOpTreeSet, cacheOpHashSet, SET_WARN_LSNR);
+
+ SET_WARN_LSNR.reset();
+
+ withTx(cacheOpHashSet, PESSIMISTIC, READ_COMMITTED);
+
+ checkOp(false, SET_WARN_LSNR);
+ }
+
+ /** */
+ private void testOp(Runnable cacheOpWithOrdered, Runnable
cacheOpWithNoOrdered, LogListener lsnr) throws Exception {
+ cacheOpWithOrdered.run();
+
+ withTx(cacheOpWithOrdered, PESSIMISTIC, SERIALIZABLE);
+
+ withTx(cacheOpWithNoOrdered, OPTIMISTIC, SERIALIZABLE);
+
+ checkOp(false, lsnr);
+
+ withTx(cacheOpWithNoOrdered, PESSIMISTIC, READ_COMMITTED);
+
+ checkOp(true, lsnr);
+ }
+
+ /** */
+ private void withTx(Runnable cacheOp, TransactionConcurrency concurrency,
TransactionIsolation isolation) {
+ try (ClientTransaction tx = cli.transactions().txStart(concurrency,
isolation)) {
+ cacheOp.run();
+
+ tx.commit();
+ }
+ }
+
+ /** */
+ private void checkOp(boolean warnPresent, LogListener lsnr) throws
Exception {
+ if (warnPresent)
+ assertTrue(waitForCondition(lsnr::check, getTestTimeout()));
Review Comment:
Use `lsnr.check(millis)`
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java:
##########
@@ -973,6 +959,75 @@ private <T> IgniteInternalFuture<Map<K,
EntryProcessorResult<T>>> invokeAll0(
});
}
+ /**
+ * @param async Async operation flag.
+ * @param invokeKeys Keys.
+ * @param entryProcessor Entry processor.
+ * @param invokeArgs Optional arguments for EntryProcessor.
+ * @return Completion future.
+ */
+ @SuppressWarnings("ConstantConditions")
+ private IgniteInternalFuture updateAll0(
Review Comment:
Let's avoid logic duplication. Instead change `updateAll0` method signature
to something like this:
```
private IgniteInternalFuture updateAll0(
@Nullable Collection<?> keys,
@Nullable Collection<? extends V> vals,
@Nullable Collection<? extends EntryProcessor> invokeVals,
@Nullable Object[] invokeArgs,
@Nullable Collection<GridCacheDrInfo> conflictPutVals,
@Nullable Collection<GridCacheVersion> conflictRmvVals,
...
```
And always pass keys parameter (keySet() of corresponding map). Also lambdas
in viewReadOnly can be simplified.
##########
modules/core/src/main/java/org/apache/ignite/internal/client/thin/TcpClientCache.java:
##########
@@ -1616,4 +1650,77 @@ private void
checkDataReplicationSupported(ProtocolContext protocolCtx)
if
(!protocolCtx.isFeatureSupported(ProtocolBitmaskFeature.DATA_REPLICATION_OPERATIONS))
throw new
ClientFeatureNotSupportedByServerException(ProtocolBitmaskFeature.DATA_REPLICATION_OPERATIONS);
}
+
+ /**
+ * Warns if an unordered map is used in an operation that may lead to a
distributed deadlock
+ * during an explicit transaction.
+ * <p>
+ * This check is relevant only for explicit user-managed transactions.
Implicit transactions
+ * (such as those started automatically by the system) are not inspected
by this method.
+ * </p>
+ *
+ * @param m The map being used in the cache operation.
+ */
+ protected void warnIfUnordered(Map<?, ?> m) {
+ if (m == null || m.size() <= 1)
+ return;
+
+ TcpClientTransaction tx = transactions.tx();
+
+ // Only explicit transactions are checked
+ if (tx == null)
+ return;
+
+ if (m instanceof SortedMap || m instanceof GridSerializableMap)
+ return;
+
+ if (!canBlockTx(false, tx.getConcurrency(), tx.getIsolation()))
+ return;
+
+ LT.warn(log, "Unordered map " + m.getClass().getName() + " is used for
putAll operation on cache " +
+ name + ". This can lead to a distributed deadlock. Switch to a
sorted map like TreeMap instead.");
+ }
+
+ /**
+ * Warns if an unordered map is used in an operation that may lead to a
distributed deadlock
+ * during an explicit transaction.
+ * <p>
+ * This check is relevant only for explicit user-managed transactions.
Implicit transactions
+ * (such as those started automatically by the system) are not inspected
by this method.
+ * </p>
+ *
+ * @param coll The collection being used in the cache operation.
+ * @param isGetOp {@code true} if the operation is a get (e.g., {@code
getAll}).
+ */
+ protected void warnIfUnordered(Collection<?> coll, boolean isGetOp) {
+ if (coll == null || coll.size() <= 1)
+ return;
+
+ TcpClientTransaction tx = transactions.tx();
+
+ // Only explicit transactions are checked
Review Comment:
Point at the end of line.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java:
##########
@@ -1004,8 +1035,8 @@ private <K, V> IgniteInternalFuture<Void> enlistWrite(
@Nullable AffinityTopologyVersion entryTopVer,
Collection<?> keys,
@Nullable ExpiryPolicy expiryPlc,
- @Nullable Map<?, ?> lookup,
- @Nullable Map<?, EntryProcessor<K, V, Object>> invokeMap,
+ @Nullable Iterator<?> itVals,
+ @Nullable Iterator<? extends EntryProcessor<K, V, Object>>
itInvokeVals,
Review Comment:
Let's use collections here and in other methods, instead of iterators:
```
@Nullable Collection<?> vals,
@Nullable Collection<? extends EntryProcessor<K, V, Object>>
invokeVals,
```
It looks more consistent (the same level of abstraction as keys) and more
safe (guaranties that nobody start iterating before, start iterating values
when we start iterating keys)
##########
modules/core/src/main/java/org/apache/ignite/internal/client/thin/TcpClientTransactions.java:
##########
@@ -106,7 +106,7 @@ private ClientTransaction txStart0(TransactionConcurrency
concurrency, Transacti
writer.writeString(lb);
}
},
- res -> new TcpClientTransaction(res.in().readInt(),
res.clientChannel())
+ res -> new TcpClientTransaction(res.in().readInt(),
res.clientChannel(), concurrency, isolation)
Review Comment:
concurrency, isolation can be null here, need to be handled using txCfg
--
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
