IGNITE-15256 Java Thin: Fix ClassNotFoundException on service call after failover (apache#9307)

Clear thin client caches when cluster connection is lost.

The root cause of the issue is that client believes that server nodes preserve registered user types after reconnect, however, it could be the case that server nodes lose this information (e.g. full cluster restart with clean up of working dirs). This leads to ClassNotFoundException.

Author: ivan-fedorenkov

modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryContext.java

@@ -1409,6 +1409,9 @@ public void unregisterUserTypeDescriptors() {
             if (e.getValue().userType())
                 it.remove();
         }
+
+        // Unregister Serializable and Externalizable type descriptors.
+        optmMarsh.clearClassDescriptorsCache();
     }
 
     /**
@@ -1438,6 +1441,8 @@ public void onUndeploy(ClassLoader ldr) {
             }
         }
 
+        optmMarsh.onUndeploy(ldr);
+
         U.clearClassCache(ldr);
     }
 

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

@@ -111,7 +111,8 @@ private TcpIgniteClient(ClientConfiguration cfg) throws ClientException {
     ) throws ClientException {
         final ClientBinaryMetadataHandler metadataHnd = new ClientBinaryMetadataHandler();
 
-        marsh = new ClientBinaryMarshaller(metadataHnd, new ClientMarshallerContext());
+        ClientMarshallerContext marshCtx = new ClientMarshallerContext();
+        marsh = new ClientBinaryMarshaller(metadataHnd, marshCtx);
 
         marsh.setBinaryConfiguration(cfg.getBinaryConfiguration());
 
@@ -124,7 +125,14 @@ private TcpIgniteClient(ClientConfiguration cfg) throws ClientException {
         try {
             ch.channelsInit();
 
-            ch.addChannelFailListener(metadataHnd::onReconnect);
+            // Metadata, binary descriptors and user types caches must be cleared so that the
+            // client will register all the user types within the cluster once again in case this information
+            // was lost during the cluster failover.
+            ch.addChannelFailListener(() -> {
+                metadataHnd.onReconnect();
+                marshCtx.clearUserTypesCache();
+                marsh.context().unregisterUserTypeDescriptors();
+            });
 
             // Send postponed metadata after channel init.
             metadataHnd.sendAllMeta();
@@ -521,7 +529,7 @@ void onReconnect() {
      */
     private class ClientMarshallerContext implements MarshallerContext {
         /** Type ID -> class name map. */
-        private Map<Integer, String> cache = new ConcurrentHashMap<>();
+        private final Map<Integer, String> cache = new ConcurrentHashMap<>();
 
         /** System types. */
         private final Collection<String> sysTypes = new HashSet<>();
@@ -646,5 +654,14 @@ public ClientMarshallerContext() {
         @Override public JdkMarshaller jdkMarshaller() {
             return new JdkMarshaller();
         }
+
+        /**
+         * Clear the user types cache on reconnect so that the client will register all
+         * the types once again after reconnect.
+         * See the comment in constructor of the TcpIgniteClient.
+         */
+        void clearUserTypesCache() {
+            cache.clear();
+        }
     }
 }

modules/core/src/main/java/org/apache/ignite/internal/marshaller/optimized/OptimizedMarshaller.java

@@ -340,6 +340,15 @@ public static boolean available() {
         U.clearClassCache(ldr);
     }
 
+    /**
+     * Clears the optimized class descriptors cache. This is essential for the clients
+     * on disconnect in order to make them register their user types again (server nodes may
+     * lose previously registered types).
+     */
+    public void clearClassDescriptorsCache() {
+        clsMap.clear();
+    }
+
     /** {@inheritDoc} */
     @Override public String toString() {
         return S.toString(OptimizedMarshaller.class, this);

modules/core/src/test/java/org/apache/ignite/client/PersonExternalizable.java

@@ -0,0 +1,58 @@
+/*
+ * 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.io.Externalizable;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import org.apache.ignite.internal.util.typedef.internal.S;
+
+/**
+ * An externalizable person entity for the tests.
+ */
+public class PersonExternalizable implements Externalizable {
+    /** */
+    private String name;
+
+    /**
+     * Externalizable
+     */
+    public PersonExternalizable() {
+    }
+
+    /** */
+    public PersonExternalizable(String name) {
+        this.name = name;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(PersonExternalizable.class, this);
+    }
+
+    /** {@inheritDoc} */
+    @Override public void writeExternal(ObjectOutput out) throws IOException {
+        out.writeUTF(name);
+    }
+
+    /** {@inheritDoc} */
+    @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+        name = in.readUTF();
+    }
+}

modules/core/src/test/java/org/apache/ignite/client/ReliabilityTest.java

@@ -37,9 +37,14 @@
 import org.apache.ignite.cache.query.Query;
 import org.apache.ignite.cache.query.QueryCursor;
 import org.apache.ignite.cache.query.ScanQuery;
+import org.apache.ignite.configuration.DataStorageConfiguration;
 import org.apache.ignite.failure.FailureHandler;
 import org.apache.ignite.internal.client.thin.AbstractThinClientTest;
 import org.apache.ignite.internal.client.thin.ClientServerError;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.services.Service;
+import org.apache.ignite.services.ServiceConfiguration;
+import org.apache.ignite.services.ServiceContext;
 import org.apache.ignite.testframework.GridTestUtils;
 import org.junit.Test;
 
@@ -51,6 +56,9 @@
  * High Availability tests.
  */
 public class ReliabilityTest extends AbstractThinClientTest {
+    /** Service name. */
+    private static final String SERVICE_NAME = "svc";
+
     /**
      * Thin clint failover.
      */
@@ -370,6 +378,69 @@ public void testServerCriticalError() throws Exception {
         }
     }
 
+    /**
+     * Test that client can invoke service method with externalizable parameter after
+     * cluster failover.
+     */
+    @Test
+    public void testServiceMethodInvocationAfterFailover() throws Exception {
+        PersonExternalizable person = new PersonExternalizable("Person 1");
+
+        ServiceConfiguration testServiceConfig = new ServiceConfiguration();
+        testServiceConfig.setName(SERVICE_NAME);
+        testServiceConfig.setService(new TestService());
+        testServiceConfig.setTotalCount(1);
+
+        Ignite ignite = null;
+        IgniteClient client = null;
+        try {
+            // Initialize cluster and client
+            ignite = startGrid(getConfiguration().setServiceConfiguration(testServiceConfig));
+            client = startClient(ignite);
+            TestServiceInterface svc = client.services().serviceProxy(SERVICE_NAME, TestServiceInterface.class);
+
+            // Invoke the service method with Externalizable parameter for the first time.
+            // This triggers registration of the PersonExternalizable type in the cluter.
+            String result = svc.testMethod(person);
+            assertEquals("testMethod(PersonExternalizable person): " + person, result);
+
+            // Kill the cluster node, clean up the working directory (with cached types)
+            // and drop the client connection.
+            ignite.close();
+            U.delete(U.resolveWorkDirectory(
+                    U.defaultWorkDirectory(),
+                    DataStorageConfiguration.DFLT_MARSHALLER_PATH,
+                    false));
+            dropAllThinClientConnections();
+
+            // Invoke the service.
+            GridTestUtils.assertThrowsWithCause(() -> svc.testMethod(person), ClientConnectionException.class);
+
+            // Restore the cluster and redeploy the service.
+            ignite = startGrid(getConfiguration().setServiceConfiguration(testServiceConfig));
+
+            // Invoke the service method with Externalizable parameter once again.
+            // This should restore the client connection and trigger registration of the
+            // PersonExternalizable once again.
+            result = svc.testMethod(person);
+            assertEquals("testMethod(PersonExternalizable person): " + person, result);
+        } finally {
+            if (ignite != null) {
+                try {
+                    ignite.close();
+                } catch (Throwable ignore) {
+                }
+            }
+
+            if (client != null) {
+                try {
+                    client.close();
+                } catch (Throwable ignore) {
+                }
+            }
+        }
+    }
+
     /**
      * Performs cache put.
      *
@@ -427,4 +498,35 @@ private void assertOnUnstableCluster(LocalIgniteCluster cluster, Runnable clo) t
     protected boolean isPartitionAware() {
         return false;
     }
+
+    /** */
+    public static interface TestServiceInterface {
+        /** */
+        public String testMethod(PersonExternalizable person);
+    }
+
+    /**
+     * Implementation of TestServiceInterface.
+     */
+    public static class TestService implements Service, TestServiceInterface {
+        /** {@inheritDoc} */
+        @Override public void cancel(ServiceContext ctx) {
+            // No-op.
+        }
+
+        /** {@inheritDoc} */
+        @Override public void init(ServiceContext ctx) throws Exception {
+            // No-op.
+        }
+
+        /** {@inheritDoc} */
+        @Override public void execute(ServiceContext ctx) throws Exception {
+            // No-op.
+        }
+
+        /** {@inheritDoc} */
+        @Override public String testMethod(PersonExternalizable person) {
+            return "testMethod(PersonExternalizable person): " + person;
+        }
+    }
 }