You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by pt...@apache.org on 2023/07/25 09:44:14 UTC
[ignite-3] branch main updated: IGNITE-19397 Return error to client when outdated schema is used (#2317)
This is an automated email from the ASF dual-hosted git repository.
ptupitsyn pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 5d44dc1617 IGNITE-19397 Return error to client when outdated schema is used (#2317)
5d44dc1617 is described below
commit 5d44dc1617958cf4e16910087f1055078bbb52e4
Author: Pavel Tupitsyn <pt...@apache.org>
AuthorDate: Tue Jul 25 12:44:09 2023 +0300
IGNITE-19397 Return error to client when outdated schema is used (#2317)
* Add `SCHEMA_VERSION_MISMATCH_ERR` to Table group
* Add internal `SchemaVersionMismatchException`
* Add error extensions to the client protocol: error details now contain an optional map with additional data
* Use error extensions to return the expected schema version to the client. The client will retry the operation using specified schema version (separate tasks)
---
.../internal/client/proto/ClientMessagePacker.java | 3 +-
.../internal/client/proto/ErrorExtensions.java} | 27 ++-------
.../client/handler/ItClientHandlerMetricsTest.java | 2 +-
.../handler/ClientInboundMessageHandler.java | 23 +++++---
.../ignite/internal/client/TcpClientChannel.java | 3 +
.../java/org/apache/ignite/lang/ErrorGroups.java | 3 +
.../dotnet/Apache.Ignite.Tests/FakeServer.cs | 1 +
.../dotnet/Apache.Ignite/Internal/ClientSocket.cs | 3 +
.../app/client/ItAbstractThinClientTest.java | 19 +++++++
.../ItThinClientSchemaSynchronizationTest.java | 58 +++++++++++++++++++
.../runner/app/client/ItThinClientSqlTest.java | 20 -------
.../ignite/internal/schema/SchemaException.java | 10 ++++
.../schema/SchemaVersionMismatchException.java | 65 ++++++++++++++++++++++
.../schema/marshaller/TupleMarshallerImpl.java | 14 +++--
14 files changed, 193 insertions(+), 58 deletions(-)
diff --git a/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientMessagePacker.java b/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientMessagePacker.java
index 9998f10174..3584b6cdad 100644
--- a/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientMessagePacker.java
+++ b/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientMessagePacker.java
@@ -27,6 +27,7 @@ import java.util.BitSet;
import java.util.UUID;
import org.apache.ignite.internal.binarytuple.BinaryTupleBuilder;
import org.apache.ignite.internal.binarytuple.BinaryTupleParser;
+import org.jetbrains.annotations.Nullable;
/**
* ByteBuf-based MsgPack implementation. Replaces {@link org.msgpack.core.MessagePacker} to avoid
@@ -270,7 +271,7 @@ public class ClientMessagePacker implements AutoCloseable {
*
* @param s the value to be written.
*/
- public void packString(String s) {
+ public void packString(@Nullable String s) {
assert !closed : "Packer is closed";
if (s == null) {
diff --git a/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaException.java b/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ErrorExtensions.java
similarity index 58%
copy from modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaException.java
copy to modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ErrorExtensions.java
index 580704f07d..f16e8b642e 100644
--- a/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaException.java
+++ b/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ErrorExtensions.java
@@ -15,30 +15,11 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.schema;
-
-import org.apache.ignite.lang.IgniteInternalException;
+package org.apache.ignite.internal.client.proto;
/**
- * Base class for schema exceptions.
+ * Error data extensions. When the server returns an error response, it may contain additional data in a map. Keys are defined here.
*/
-public class SchemaException extends IgniteInternalException {
- /**
- * Constructor with error message.
- *
- * @param msg Message.
- */
- public SchemaException(String msg) {
- super(msg);
- }
-
- /**
- * Constructor with error message and cause.
- *
- * @param msg Message.
- * @param cause Cause.
- */
- public SchemaException(String msg, Throwable cause) {
- super(msg, cause);
- }
+public class ErrorExtensions {
+ public static final String EXPECTED_SCHEMA_VERSION = "expected-schema-ver";
}
diff --git a/modules/client-handler/src/integrationTest/java/org/apache/ignite/client/handler/ItClientHandlerMetricsTest.java b/modules/client-handler/src/integrationTest/java/org/apache/ignite/client/handler/ItClientHandlerMetricsTest.java
index 62e0e67e13..537025c58e 100644
--- a/modules/client-handler/src/integrationTest/java/org/apache/ignite/client/handler/ItClientHandlerMetricsTest.java
+++ b/modules/client-handler/src/integrationTest/java/org/apache/ignite/client/handler/ItClientHandlerMetricsTest.java
@@ -147,7 +147,7 @@ public class ItClientHandlerMetricsTest {
ItClientHandlerTestUtils.connectAndHandshake(serverModule, false, true);
assertTrue(
- IgniteTestUtils.waitForCondition(() -> testServer.metrics().bytesSent() == 158, 1000),
+ IgniteTestUtils.waitForCondition(() -> testServer.metrics().bytesSent() == 159, 1000),
() -> "bytesSent: " + testServer.metrics().bytesSent());
assertTrue(
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java
index 04f5058eda..f41ebf8e8c 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java
@@ -86,6 +86,7 @@ import org.apache.ignite.internal.client.proto.ClientMessageCommon;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.client.proto.ClientOp;
+import org.apache.ignite.internal.client.proto.ErrorExtensions;
import org.apache.ignite.internal.client.proto.HandshakeExtension;
import org.apache.ignite.internal.client.proto.ProtocolVersion;
import org.apache.ignite.internal.client.proto.ResponseFlags;
@@ -95,6 +96,7 @@ import org.apache.ignite.internal.jdbc.proto.JdbcQueryCursorHandler;
import org.apache.ignite.internal.jdbc.proto.JdbcQueryEventHandler;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.schema.SchemaVersionMismatchException;
import org.apache.ignite.internal.security.authentication.AnonymousRequest;
import org.apache.ignite.internal.security.authentication.AuthenticationManager;
import org.apache.ignite.internal.security.authentication.AuthenticationRequest;
@@ -396,6 +398,7 @@ public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter im
private void writeErrorCore(Throwable err, ClientMessagePacker packer) {
err = ExceptionUtils.unwrapCause(err);
+ // Trace ID and error code.
if (err instanceof TraceableException) {
TraceableException iex = (TraceableException) err;
packer.packUuid(iex.traceId());
@@ -405,21 +408,25 @@ public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter im
packer.packInt(INTERNAL_ERR);
}
+ // Class name and message.
packer.packString(err.getClass().getName());
+ packer.packString(err.getMessage());
- String msg = err.getMessage();
-
- if (msg == null) {
- packer.packNil();
- } else {
- packer.packString(msg);
- }
-
+ // Stack trace.
if (configuration.sendServerExceptionStackTraceToClient()) {
packer.packString(ExceptionUtils.getFullStackTrace(err));
} else {
packer.packNil();
}
+
+ // Extensions.
+ if (err instanceof SchemaVersionMismatchException) {
+ packer.packMapHeader(1);
+ packer.packString(ErrorExtensions.EXPECTED_SCHEMA_VERSION);
+ packer.packInt(((SchemaVersionMismatchException) err).expectedVersion());
+ } else {
+ packer.packNil(); // No extensions.
+ }
}
private static ClientMessagePacker getPacker(ByteBufAllocator alloc) {
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java b/modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java
index edeaca4976..fc415dbebf 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java
@@ -432,6 +432,9 @@ class TcpClientChannel implements ClientChannel, ClientMessageHandler, ClientCon
IgniteException causeWithStackTrace = unpacker.tryUnpackNil() ? null : new IgniteException(traceId, code, unpacker.unpackString());
+ // TODO IGNITE-19837 Retry outdated schema error
+ unpacker.skipValues(1); // Error extensions.
+
try {
// TODO https://issues.apache.org/jira/browse/IGNITE-19539
Class<? extends Throwable> errCls = (Class<? extends Throwable>) Class.forName(errClassName);
diff --git a/modules/core/src/main/java/org/apache/ignite/lang/ErrorGroups.java b/modules/core/src/main/java/org/apache/ignite/lang/ErrorGroups.java
index d41145d69b..50f85f0049 100755
--- a/modules/core/src/main/java/org/apache/ignite/lang/ErrorGroups.java
+++ b/modules/core/src/main/java/org/apache/ignite/lang/ErrorGroups.java
@@ -71,6 +71,9 @@ public class ErrorGroups {
/** Table definition is incorrect. */
public static final int TABLE_DEFINITION_ERR = TABLE_ERR_GROUP.registerErrorCode((short) 6);
+
+ /** Schema version mismatch. */
+ public static final int SCHEMA_VERSION_MISMATCH_ERR = TABLE_ERR_GROUP.registerErrorCode((short) 7);
}
/** Client error group. */
diff --git a/modules/platforms/dotnet/Apache.Ignite.Tests/FakeServer.cs b/modules/platforms/dotnet/Apache.Ignite.Tests/FakeServer.cs
index ce9525b91f..96dadabeff 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Tests/FakeServer.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Tests/FakeServer.cs
@@ -320,6 +320,7 @@ namespace Apache.Ignite.Tests
w.Write("org.foo.bar.BazException");
w.Write(Err);
w.WriteNil(); // Stack trace.
+ w.WriteNil(); // Error extensions.
Send(handler, requestId, errWriter, isError: true);
}
diff --git a/modules/platforms/dotnet/Apache.Ignite/Internal/ClientSocket.cs b/modules/platforms/dotnet/Apache.Ignite/Internal/ClientSocket.cs
index 135dc74893..f9319fc3d7 100644
--- a/modules/platforms/dotnet/Apache.Ignite/Internal/ClientSocket.cs
+++ b/modules/platforms/dotnet/Apache.Ignite/Internal/ClientSocket.cs
@@ -400,6 +400,9 @@ namespace Apache.Ignite.Internal
string? message = reader.ReadStringNullable();
string? javaStackTrace = reader.ReadStringNullable();
+ // TODO IGNITE-19838 Retry outdated schema error
+ reader.Skip(); // Error extensions.
+
return ExceptionMapper.GetException(traceId, code, className, message, javaStackTrace);
}
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItAbstractThinClientTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItAbstractThinClientTest.java
index b0d071a8b2..6cc30ae373 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItAbstractThinClientTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItAbstractThinClientTest.java
@@ -184,6 +184,25 @@ public abstract class ItAbstractThinClientTest extends IgniteAbstractTest {
return startedNodes.get(idx);
}
+ void waitForTableOnAllNodes(String tableName) throws InterruptedException {
+ // TODO IGNITE-18733, IGNITE-18449: remove this workaround when issues are fixed.
+ // Currently newly created table is not immediately available on all nodes.
+ boolean res = IgniteTestUtils.waitForCondition(() -> {
+ var nodeCount = client().clusterNodes().size();
+
+ // Do N checks - they will go to different nodes becase of request balancing.
+ for (int i = 0; i < nodeCount; i++) {
+ if (client().tables().table(tableName) == null) {
+ return false;
+ }
+ }
+
+ return true;
+ }, 3000);
+
+ assertTrue(res);
+ }
+
/**
* Test class.
*/
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientSchemaSynchronizationTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientSchemaSynchronizationTest.java
new file mode 100644
index 0000000000..1be0a64acd
--- /dev/null
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientSchemaSynchronizationTest.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.internal.runner.app.client;
+
+import static org.hamcrest.CoreMatchers.containsString;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.lang.IgniteException;
+import org.apache.ignite.sql.Session;
+import org.apache.ignite.table.RecordView;
+import org.apache.ignite.table.Tuple;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for client schema synchronization.
+ */
+public class ItThinClientSchemaSynchronizationTest extends ItAbstractThinClientTest {
+ @SuppressWarnings("resource")
+ @Test
+ void testOutdatedSchemaFromClientThrowsExceptionOnServer() throws InterruptedException {
+ IgniteClient client = client();
+ Session ses = client.sql().createSession();
+
+ // Create table, insert data.
+ String tableName = "testOutdatedSchemaFromClientThrowsExceptionOnServer";
+ ses.execute(null, "CREATE TABLE " + tableName + "(ID INT NOT NULL PRIMARY KEY)");
+
+ waitForTableOnAllNodes(tableName);
+ RecordView<Tuple> recordView = client.tables().table(tableName).recordView();
+
+ Tuple rec = Tuple.create().set("ID", 1);
+ recordView.insert(null, rec);
+
+ // Modify table, get data - client will use old schema.
+ ses.execute(null, "ALTER TABLE testOutdatedSchemaFromClientThrowsExceptionOnServer ADD COLUMN NAME VARCHAR");
+
+ // TODO IGNITE-19837 Retry outdated schema error
+ IgniteException ex = assertThrows(IgniteException.class, () -> recordView.insert(null, rec));
+ assertThat(ex.getMessage(), containsString("Schema version mismatch [expectedVer=2, actualVer=1]"));
+ }
+}
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientSqlTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientSqlTest.java
index 3f82661caf..835c3dc292 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientSqlTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientSqlTest.java
@@ -34,7 +34,6 @@ import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CompletionException;
-import org.apache.ignite.internal.testframework.IgniteTestUtils;
import org.apache.ignite.lang.IgniteException;
import org.apache.ignite.sql.ColumnMetadata;
import org.apache.ignite.sql.ColumnType;
@@ -516,25 +515,6 @@ public class ItThinClientSqlTest extends ItAbstractThinClientTest {
assertEquals(ColumnType.BYTE_ARRAY, meta.columns().get(13).type());
}
- private void waitForTableOnAllNodes(String tableName) throws InterruptedException {
- // TODO IGNITE-18733, IGNITE-18449: remove this workaround when issues are fixed.
- // Currently newly created table is not immediately available on all nodes.
- boolean res = IgniteTestUtils.waitForCondition(() -> {
- var nodeCount = client().clusterNodes().size();
-
- // Do N checks - they will go to different nodes becase of request balancing.
- for (int i = 0; i < nodeCount; i++) {
- if (client().tables().table(tableName) == null) {
- return false;
- }
- }
-
- return true;
- }, 3000);
-
- assertTrue(res);
- }
-
private static class Pojo {
public long num;
diff --git a/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaException.java b/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaException.java
index 580704f07d..b6cb9cbc65 100644
--- a/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaException.java
+++ b/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaException.java
@@ -41,4 +41,14 @@ public class SchemaException extends IgniteInternalException {
public SchemaException(String msg, Throwable cause) {
super(msg, cause);
}
+
+ /**
+ * Creates a new exception with the given error code and detail message.
+ *
+ * @param code Full error code.
+ * @param message Detail message.
+ */
+ public SchemaException(int code, String message) {
+ super(code, message);
+ }
}
diff --git a/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaVersionMismatchException.java b/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaVersionMismatchException.java
new file mode 100644
index 0000000000..0f3444576a
--- /dev/null
+++ b/modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaVersionMismatchException.java
@@ -0,0 +1,65 @@
+/*
+ * 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.schema;
+
+import static org.apache.ignite.lang.ErrorGroups.Table.SCHEMA_VERSION_MISMATCH_ERR;
+
+/**
+ * Indicates incompatible schema version.
+ */
+public final class SchemaVersionMismatchException extends SchemaException {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Expected schema version. */
+ private final int expectedVersion;
+
+ /** Actual schema version. */
+ private final int actualVersion;
+
+ /**
+ * Constructor.
+ *
+ * @param expectedVersion Expected schema version.
+ * @param actualVer Actual schema version.
+ */
+ public SchemaVersionMismatchException(int expectedVersion, int actualVer) {
+ super(SCHEMA_VERSION_MISMATCH_ERR, "Schema version mismatch [expectedVer=" + expectedVersion + ", actualVer=" + actualVer + ']');
+
+ this.expectedVersion = expectedVersion;
+ this.actualVersion = actualVer;
+ }
+
+ /**
+ * Gets expected schema version.
+ *
+ * @return Expected schema version.
+ */
+ public int expectedVersion() {
+ return expectedVersion;
+ }
+
+ /**
+ * Gets actual schema version.
+ *
+ * @return Actual schema version.
+ */
+ public int actualVersion() {
+ return actualVersion;
+ }
+}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/schema/marshaller/TupleMarshallerImpl.java b/modules/table/src/main/java/org/apache/ignite/internal/schema/marshaller/TupleMarshallerImpl.java
index 0eed13cdb3..2f76d458fa 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/schema/marshaller/TupleMarshallerImpl.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/schema/marshaller/TupleMarshallerImpl.java
@@ -33,6 +33,7 @@ import org.apache.ignite.internal.schema.SchemaAware;
import org.apache.ignite.internal.schema.SchemaDescriptor;
import org.apache.ignite.internal.schema.SchemaMismatchException;
import org.apache.ignite.internal.schema.SchemaRegistry;
+import org.apache.ignite.internal.schema.SchemaVersionMismatchException;
import org.apache.ignite.internal.schema.row.Row;
import org.apache.ignite.internal.schema.row.RowAssembler;
import org.apache.ignite.table.Tuple;
@@ -70,11 +71,14 @@ public class TupleMarshallerImpl implements TupleMarshaller {
SchemaDescriptor tupleSchema = ((SchemaAware) tuple).schema();
BinaryTupleReader tupleReader = ((BinaryTupleContainer) tuple).binaryTuple();
- if (tupleSchema != null
- && tupleReader != null
- && tupleSchema.version() == schema.version()
- && !binaryTupleRebuildRequired(schema)) {
- return new Row(schema, RowAssembler.build(tupleReader.byteBuffer(), schema.version(), true));
+ if (tupleSchema != null && tupleReader != null) {
+ if (tupleSchema.version() != schema.version()) {
+ throw new SchemaVersionMismatchException(schema.version(), tupleSchema.version());
+ }
+
+ if (!binaryTupleRebuildRequired(schema)) {
+ return new Row(schema, RowAssembler.build(tupleReader.byteBuffer(), schema.version(), true));
+ }
}
}