IGNITE-27829 Java thin: Add cache name to exceptions with cache opera…

modules/core/src/main/java/org/apache/ignite/internal/client/thin/ClientUtils.java

@@ -48,6 +48,7 @@
 import org.apache.ignite.cache.QueryIndexType;
 import org.apache.ignite.cache.query.SqlFieldsQuery;
 import org.apache.ignite.client.ClientCacheConfiguration;
+import org.apache.ignite.client.ClientFeatureNotSupportedByServerException;
 import org.apache.ignite.internal.binary.BinaryContext;
 import org.apache.ignite.internal.binary.BinaryFieldMetadata;
 import org.apache.ignite.internal.binary.BinaryMetadata;
@@ -369,16 +370,17 @@ void cacheConfiguration(ClientCacheConfiguration cfg, boolean sql, BinaryOutputS
                 });
             }
             else if (cfg.getExpiryPolicy() != null) {
-                throw new ClientProtocolError(String.format("Expire policies are not supported by the server " +
-                    "version %s, required version %s", protocolCtx.version(), EXPIRY_POLICY.verIntroduced()));
+                throw new ClientFeatureNotSupportedByServerException(String.format(
+                    "Expire policies are not supported by the server version %s, required version %s",
+                    protocolCtx.version(), EXPIRY_POLICY.verIntroduced()));
             }
 
             if (protocolCtx.isFeatureSupported(CACHE_STORAGES)) {
                 itemWriter.accept(CfgItem.STORAGE_PATH, w -> w.writeStringArray(cfg.getStoragePaths()));
                 itemWriter.accept(CfgItem.IDX_PATH, w -> w.writeString(cfg.getIndexPath()));
             }
             else if (!F.isEmpty(cfg.getStoragePaths()) || !F.isEmpty(cfg.getIndexPath()))
-                throw new ClientProtocolError("Cache storages are not supported by the server");
+                throw new ClientFeatureNotSupportedByServerException("Cache storages are not supported by the server");
 
             writer.writeInt(origPos, out.position() - origPos - 4); // configuration length
             writer.writeInt(origPos + 4, propCnt.get()); // properties count

modules/core/src/main/java/org/apache/ignite/internal/client/thin/TcpClientCache.java

@@ -26,6 +26,7 @@
 import java.util.Set;
 import java.util.SortedMap;
 import java.util.SortedSet;
+import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -108,7 +109,7 @@ public class TcpClientCache<K, V> implements ClientCache<K, V> {
     private final int cacheId;
 
     /** Channel. */
-    private final ReliableChannelImpl ch;
+    private final ReliableChannelEx ch;
 
     /** Cache name. */
     private final String name;
@@ -144,11 +145,13 @@ public class TcpClientCache<K, V> implements ClientCache<K, V> {
     /** Constructor. */
     TcpClientCache(String name, ReliableChannelImpl ch, ClientBinaryMarshaller marsh, TcpClientTransactions transactions,
         ClientCacheEntryListenersRegistry lsnrsRegistry, IgniteLogger log) {
-        this(name, ch, marsh, transactions, lsnrsRegistry, false, null, log);
+        this(name, new ReliableChannelWrapper(ch, name), marsh, transactions, lsnrsRegistry, false, null, log);
+
+        ch.registerCacheIfCustomAffinity(name);
     }
 
     /** Constructor. */
-    TcpClientCache(String name, ReliableChannelImpl ch, ClientBinaryMarshaller marsh,
+    private TcpClientCache(String name, ReliableChannelEx ch, ClientBinaryMarshaller marsh,
         TcpClientTransactions transactions, ClientCacheEntryListenersRegistry lsnrsRegistry, boolean keepBinary,
         ExpiryPolicy expiryPlc, IgniteLogger log) {
         this.name = name;
@@ -165,8 +168,6 @@ public class TcpClientCache<K, V> implements ClientCache<K, V> {
 
         jCacheAdapter = new ClientJCacheAdapter<>(this);
 
-        this.ch.registerCacheIfCustomAffinity(this.name);
-
         this.log = log;
     }
 
@@ -1428,6 +1429,9 @@ private <T> T txAwareService(
                 throw new ClientException("Transaction context has been lost due to connection errors. " +
                     "Cache operations are prohibited until current transaction closed.", e);
             }
+            catch (Exception e) {
+                throw convertException(e, name);
+            }
         }
         else if (affKey != null)
             return ch.affinityService(cacheId, affKey, op, payloadWriter, payloadReader);
@@ -1459,7 +1463,7 @@ private <T> IgniteClientFuture<T> txAwareServiceAsync(
                             "Cache operations are prohibited until current transaction closed.", err));
                 }
                 else if (err != null)
-                    fut.completeExceptionally(err);
+                    fut.completeExceptionally(convertException((Exception)err, name));
                 else
                     fut.complete(res);
             });
@@ -1540,8 +1544,9 @@ private void writeCacheInfo(PayloadOutputChannel payloadCh, TcpClientTransaction
             ProtocolContext protocolCtx = payloadCh.clientChannel().protocolCtx();
 
             if (!protocolCtx.isFeatureSupported(EXPIRY_POLICY)) {
-                throw new ClientProtocolError(String.format("Expire policies are not supported by the server " +
-                    "version %s, required version %s", protocolCtx.version(), EXPIRY_POLICY.verIntroduced()));
+                throw new ClientFeatureNotSupportedByServerException(String.format(
+                    "Expire policies are not supported by the server version %s, required version %s",
+                    protocolCtx.version(), EXPIRY_POLICY.verIntroduced()));
             }
 
             flags |= WITH_EXPIRY_POLICY_FLAG_MASK;
@@ -1754,4 +1759,136 @@ private boolean canBlockTx(boolean isGetOp, TransactionConcurrency concurrency,
 
         return true;
     }
+
+    /** */
+    private static ClientException convertException(Exception e, String cacheName) {
+        String msg = "Failed to perform cache operation [cacheName=" + cacheName + "]: " + e.getMessage();
+
+        // Exception can be ClientProtocolError - it's an internal exception, can't be thrown to user.
+        if (!(e instanceof ClientException))
+            return new ClientException(msg, e);
+        else if (X.hasCause(e, ClientServerError.class)) // Wrap server errors.
+            return new ClientException(msg, e);
+        else // Don't wrap authentication, authorization, connection errors.
+            return (ClientException)e;
+    }
+
+    /** */
+    private static class ReliableChannelWrapper implements ReliableChannelEx {
+        /** */
+        private final ReliableChannelImpl delegate;
+
+        /** */
+        private final String cacheName;
+
+        /** */
+        public ReliableChannelWrapper(ReliableChannelImpl delegate, String cacheName) {
+            this.delegate = delegate;
+            this.cacheName = cacheName;
+        }
+
+        /** {@inheritDoc} */
+        @Override public <T> T service(
+            ClientOperation op,
+            Consumer<PayloadOutputChannel> payloadWriter,
+            Function<PayloadInputChannel, T> payloadReader
+        ) throws ClientException, ClientError {
+            try {
+                return delegate.service(op, payloadWriter, payloadReader);
+            }
+            catch (ClientException e) {
+                throw convertException(e, cacheName);
+            }
+        }
+
+        /** {@inheritDoc} */
+        @Override public <T> T service(
+            ClientOperation op,
+            Consumer<PayloadOutputChannel> payloadWriter,
+            Function<PayloadInputChannel, T> payloadReader,
+            List<UUID> targetNodes
+        ) throws ClientException, ClientError {
+            try {
+                return delegate.service(op, payloadWriter, payloadReader, targetNodes);
+            }
+            catch (ClientException e) {
+                throw convertException(e, cacheName);
+            }
+        }
+
+        /** {@inheritDoc} */
+        @Override public <T> IgniteClientFuture<T> serviceAsync(
+            ClientOperation op,
+            Consumer<PayloadOutputChannel> payloadWriter,
+            Function<PayloadInputChannel, T> payloadReader
+        ) throws ClientException, ClientError {
+            CompletableFuture<T> fut = new CompletableFuture<>();
+
+            delegate.serviceAsync(op, payloadWriter, payloadReader).whenComplete((res, err) -> {
+                if (err != null)
+                    fut.completeExceptionally(convertException((Exception)err, cacheName));
+                else
+                    fut.complete(res);
+            });
+
+            return new IgniteClientFutureImpl<>(fut);
+        }
+
+        /** {@inheritDoc} */
+        @Override public <T> T affinityService(
+            int cacheId,
+            Object key,
+            ClientOperation op,
+            Consumer<PayloadOutputChannel> payloadWriter,
+            Function<PayloadInputChannel, T> payloadReader
+        ) throws ClientException, ClientError {
+            try {
+                return delegate.affinityService(cacheId, key, op, payloadWriter, payloadReader);
+            }
+            catch (ClientException e) {
+                throw convertException(e, cacheName);
+            }
+        }
+
+        /** {@inheritDoc} */
+        @Override public <T> T affinityService(
+            int cacheId,
+            int part,
+            ClientOperation op,
+            Consumer<PayloadOutputChannel> payloadWriter,
+            Function<PayloadInputChannel, T> payloadReader
+        ) throws ClientException, ClientError {
+            try {
+                return delegate.affinityService(cacheId, part, op, payloadWriter, payloadReader);
+            }
+            catch (ClientException e) {
+                throw convertException(e, cacheName);
+            }
+        }
+
+        /** {@inheritDoc} */
+        @Override public <T> IgniteClientFuture<T> affinityServiceAsync(
+            int cacheId,
+            Object key,
+            ClientOperation op,
+            Consumer<PayloadOutputChannel> payloadWriter,
+            Function<PayloadInputChannel, T> payloadReader
+        ) throws ClientException, ClientError {
+            CompletableFuture<T> fut = new CompletableFuture<>();
+
+            delegate.affinityServiceAsync(cacheId, key, op, payloadWriter, payloadReader).whenComplete((res, err) -> {
+                if (err != null)
+                    fut.completeExceptionally(convertException((Exception)err, cacheName));
+                else
+                    fut.complete(res);
+            });
+
+            return new IgniteClientFutureImpl<>(fut);
+        }
+
+        /** {@inheritDoc} */
+        @Override public void close() {
+            delegate.close();
+        }
+    }
 }

modules/core/src/main/java/org/apache/ignite/internal/client/thin/TcpClientTransactions.java

@@ -22,6 +22,7 @@
 import java.util.concurrent.atomic.AtomicLong;
 import org.apache.ignite.client.ClientConnectionException;
 import org.apache.ignite.client.ClientException;
+import org.apache.ignite.client.ClientFeatureNotSupportedByServerException;
 import org.apache.ignite.client.ClientTransaction;
 import org.apache.ignite.client.ClientTransactions;
 import org.apache.ignite.configuration.ClientTransactionConfiguration;
@@ -95,8 +96,9 @@ private ClientTransaction txStart0(TransactionConcurrency concurrency, Transacti
                 ProtocolContext protocolCtx = req.clientChannel().protocolCtx();
 
                 if (!protocolCtx.isFeatureSupported(TRANSACTIONS)) {
-                    throw new ClientProtocolError(String.format("Transactions are not supported by the server's " +
-                        "protocol version %s, required version %s", protocolCtx.version(), TRANSACTIONS.verIntroduced()));
+                    throw new ClientFeatureNotSupportedByServerException(String.format(
+                        "Transactions are not supported by the server's protocol version %s, required version %s",
+                        protocolCtx.version(), TRANSACTIONS.verIntroduced()));
                 }
 
                 try (BinaryWriterEx writer = BinaryUtils.writer(marsh.context(), req.out(), null)) {

modules/core/src/test/java/org/apache/ignite/internal/client/thin/CacheExceptionsTest.java

@@ -0,0 +1,84 @@
+/*
+ * 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.client.thin;
+
+import java.util.Objects;
+import org.apache.ignite.cache.query.ScanQuery;
+import org.apache.ignite.client.ClientCache;
+import org.apache.ignite.client.ClientException;
+import org.apache.ignite.client.ClientTransaction;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.junit.Test;
+
+/**
+ * Thin client cache exceptions tests.
+ */
+public class CacheExceptionsTest extends AbstractThinClientTest {
+    /** {@inheritDoc} */
+    @Override protected void beforeTestsStarted() throws Exception {
+        super.beforeTestsStarted();
+
+        startGrid(0);
+    }
+
+    /**
+     * Tests cache name in wrapped cache exception for server errors.
+     */
+    @Test
+    public void testCacheExceptionWrapped() throws Exception {
+        try (IgniteClient client = startClient(0)) {
+            String cacheName = "testCacheName";
+
+            ClientCache<Object, Objects> cache = client.cache(cacheName);
+
+            // Affinity call.
+            GridTestUtils.assertThrowsAnyCause(log, () -> cache.get(0), ClientException.class, cacheName);
+
+            // Async affinity call.
+            GridTestUtils.assertThrowsAnyCause(log, () -> cache.getAsync(0).get(), ClientException.class, cacheName);
+
+            // Non-affinity call.
+            GridTestUtils.assertThrowsAnyCause(log, () -> cache.size(), ClientException.class, cacheName);
+
+            // Async non-affinity call.
+            GridTestUtils.assertThrowsAnyCause(log, () -> cache.sizeAsync().get(), ClientException.class, cacheName);
+
+            // Transactional call.
+            GridTestUtils.assertThrowsAnyCause(log, () -> {
+                try (ClientTransaction ignore = client.transactions().txStart()) {
+                    return cache.get(0);
+                }
+            }, ClientException.class, cacheName);
+
+            // Async transactional call.
+            GridTestUtils.assertThrowsAnyCause(log, () -> {
+                try (ClientTransaction ignore = client.transactions().txStart()) {
+                    return cache.getAsync(0).get();
+                }
+            }, ClientException.class, cacheName);
+
+            // Query.
+            GridTestUtils.assertThrowsAnyCause(log, () -> cache.query(new ScanQuery<>()).getAll(), ClientException.class, cacheName);
+
+            // Affinity query.
+            GridTestUtils.assertThrowsAnyCause(log, () -> cache.query(new ScanQuery<>().setPartition(0)).getAll(),
+                ClientException.class, cacheName);
+        }
+    }
+}

modules/core/src/test/java/org/apache/ignite/internal/client/thin/events/IgniteClientRequestEventListenerTest.java

@@ -31,6 +31,7 @@
 import org.apache.ignite.configuration.ClientConfiguration;
 import org.apache.ignite.internal.client.thin.AbstractThinClientTest;
 import org.apache.ignite.internal.client.thin.ClientOperation;
+import org.apache.ignite.internal.util.typedef.X;
 import org.junit.Test;
 
 /**
@@ -154,7 +155,7 @@ public void testQueryFailEvents() {
             assertEquals(ClientOperation.CACHE_PUT.code(), startEvt.operationCode());
             assertEquals(ClientOperation.CACHE_PUT.name(), startEvt.operationName());
 
-            assertEquals(err, failEvt.throwable());
+            assertTrue(X.getThrowableList(err).contains(failEvt.throwable()));
 
             assertTrue(System.nanoTime() - startTime >= failEvt.elapsedTime(TimeUnit.NANOSECONDS));
         }

modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java

@@ -1671,19 +1671,6 @@ protected <T extends Event> IgniteFuture<T> waitForLocalEvent(IgniteEvents evts,
         return fut;
     }
 
-    /**
-     * @param e Exception.
-     * @param exCls Ex class.
-     */
-    protected <T extends IgniteException> void assertCacheExceptionWithCause(RuntimeException e, Class<T> exCls) {
-        if (exCls.isAssignableFrom(e.getClass()))
-            return;
-
-        if (e.getClass() != CacheException.class
-            || e.getCause() == null || !exCls.isAssignableFrom(e.getCause().getClass()))
-            throw e;
-    }
-
     /**
      * @param cache Cache.
      */

modules/indexing/src/test/java/org/apache/ignite/client/ClientTestSuite.java

@@ -23,6 +23,7 @@
 import org.apache.ignite.internal.client.thin.BlockingTxOpsTest;
 import org.apache.ignite.internal.client.thin.CacheAsyncTest;
 import org.apache.ignite.internal.client.thin.CacheEntryListenersTest;
+import org.apache.ignite.internal.client.thin.CacheExceptionsTest;
 import org.apache.ignite.internal.client.thin.ClusterApiTest;
 import org.apache.ignite.internal.client.thin.ClusterGroupClusterRestartTest;
 import org.apache.ignite.internal.client.thin.ClusterGroupTest;
@@ -108,7 +109,8 @@
     InvokeTest.class,
     ExtraColumnInH2RowsTest.class,
     RecoveryModeTest.class,
-    ReliableChannelDuplicationTest.class
+    ReliableChannelDuplicationTest.class,
+    CacheExceptionsTest.class,
 })
 public class ClientTestSuite {
     // No-op.