/** {@inheritDoc} */
@Override public boolean registerClassName(
- byte platformId,
- int typeId,
- String clsName
+ byte platformId,
+ int typeId,
+ String clsName,
+ boolean failIfUnregistered
) throws IgniteCheckedException {
ConcurrentMap<Integer, MappedName> cache = getCacheFor(platformId);
if (transport.stopping())
return false;
- IgniteInternalFuture<MappingExchangeResult> fut = transport.awaitMappingAcceptance(new MarshallerMappingItem(platformId, typeId, clsName), cache);
+ MarshallerMappingItem item = new MarshallerMappingItem(platformId, typeId, clsName);
+
+ GridFutureAdapter<MappingExchangeResult> fut = transport.awaitMappingAcceptance(item, cache);
+
+ if (failIfUnregistered && !fut.isDone())
+ throw new UnregisteredBinaryTypeException(typeId, fut);
+
MappingExchangeResult res = fut.get();
return convertXchRes(res);
if (transport.stopping())
return false;
- IgniteInternalFuture<MappingExchangeResult> fut = transport.proposeMapping(new MarshallerMappingItem(platformId, typeId, clsName), cache);
+ MarshallerMappingItem item = new MarshallerMappingItem(platformId, typeId, clsName);
+
+ GridFutureAdapter<MappingExchangeResult> fut = transport.proposeMapping(item, cache);
+
+ if (failIfUnregistered && !fut.isDone())
+ throw new UnregisteredBinaryTypeException(typeId, fut);
+
MappingExchangeResult res = fut.get();
return convertXchRes(res);
}
/** {@inheritDoc} */
+ @Override
+ public boolean registerClassName(byte platformId, int typeId, String clsName) {
+ throw new UnsupportedOperationException("registerClassName");
+ }
+
+ /** {@inheritDoc} */
@Override public boolean registerClassNameLocally(byte platformId, int typeId, String clsName)
throws IgniteCheckedException
{
import org.apache.ignite.IgniteException;
import org.apache.ignite.internal.binary.BinaryMetadata;
-import org.jetbrains.annotations.Nullable;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
/**
* Exception thrown during serialization if binary metadata isn't registered and it's registration isn't allowed.
+ * Used for both binary types and marshalling mappings.
+ * Confusing old class name is preserved for backwards compatibility.
*/
public class UnregisteredBinaryTypeException extends IgniteException {
/** */
private static final long serialVersionUID = 0L;
/** */
+ private static final String MESSAGE =
+ "Attempted to update binary metadata inside a critical synchronization block (will be " +
+ "automatically retried). This exception must not be wrapped to any other exception class. " +
+ "If you encounter this exception outside of EntryProcessor, please report to Apache Ignite " +
+ "dev-list. Debug info [typeId=%d, binaryMetadata=%s, fut=%s]";
+
+ /** */
+ private static String createMessage(int typeId, BinaryMetadata binaryMetadata, GridFutureAdapter<?> fut) {
+ return String.format(MESSAGE, typeId, binaryMetadata, fut);
+ }
+
+ /** */
private final int typeId;
/** */
private final BinaryMetadata binaryMetadata;
+ /** */
+ private final GridFutureAdapter<?> fut;
+
/**
* @param typeId Type ID.
* @param binaryMetadata Binary metadata.
*/
public UnregisteredBinaryTypeException(int typeId, BinaryMetadata binaryMetadata) {
- this.typeId = typeId;
- this.binaryMetadata = binaryMetadata;
+ this(typeId, binaryMetadata, null);
}
/**
- * @param msg Error message.
* @param typeId Type ID.
- * @param binaryMetadata Binary metadata.
+ * @param fut Future to wait in handler.
*/
- public UnregisteredBinaryTypeException(String msg, int typeId,
- BinaryMetadata binaryMetadata) {
- super(msg);
- this.typeId = typeId;
- this.binaryMetadata = binaryMetadata;
+ public UnregisteredBinaryTypeException(int typeId, GridFutureAdapter<?> fut) {
+ this(typeId, null, fut);
}
/**
- * @param cause Non-null throwable cause.
* @param typeId Type ID.
* @param binaryMetadata Binary metadata.
+ * @param fut Future to wait in handler.
*/
- public UnregisteredBinaryTypeException(Throwable cause, int typeId,
- BinaryMetadata binaryMetadata) {
- super(cause);
- this.typeId = typeId;
- this.binaryMetadata = binaryMetadata;
- }
+ private UnregisteredBinaryTypeException(int typeId, BinaryMetadata binaryMetadata, GridFutureAdapter<?> fut) {
+ super(createMessage(typeId, binaryMetadata, fut));
- /**
- * @param msg Error message.
- * @param cause Non-null throwable cause.
- * @param typeId Type ID.
- * @param binaryMetadata Binary metadata.
- */
- public UnregisteredBinaryTypeException(String msg, @Nullable Throwable cause, int typeId,
- BinaryMetadata binaryMetadata) {
- super(msg, cause);
this.typeId = typeId;
this.binaryMetadata = binaryMetadata;
+ this.fut = fut;
}
/**
public BinaryMetadata binaryMetadata() {
return binaryMetadata;
}
+
+ /**
+ * @return Future to wait in handler.
+ */
+ public GridFutureAdapter<?> future() {
+ return fut;
+ }
}
import org.apache.ignite.binary.BinaryReflectiveSerializer;
import org.apache.ignite.binary.BinarySerializer;
import org.apache.ignite.binary.Binarylizable;
-import org.apache.ignite.internal.UnregisteredClassException;
import org.apache.ignite.internal.UnregisteredBinaryTypeException;
+import org.apache.ignite.internal.UnregisteredClassException;
import org.apache.ignite.internal.marshaller.optimized.OptimizedMarshaller;
import org.apache.ignite.internal.processors.cache.CacheObjectImpl;
import org.apache.ignite.internal.processors.query.QueryUtils;
assert false : "Invalid mode: " + mode;
}
}
+ catch (UnregisteredBinaryTypeException | UnregisteredClassException e) {
+ throw e;
+ }
catch (Exception e) {
- if (e instanceof UnregisteredBinaryTypeException || e instanceof UnregisteredClassException)
- throw e;
-
String msg;
if (S.INCLUDE_SENSITIVE && !F.isEmpty(typeName))
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.internal.UnregisteredBinaryTypeException;
import org.apache.ignite.internal.UnregisteredClassException;
+import org.apache.ignite.internal.processors.marshaller.MappingExchangeResult;
import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.binary.BinaryBasicIdMapper;
import org.apache.ignite.binary.BinaryBasicNameMapper;
final int typeId = mapper.typeId(clsName);
- registered = registerUserClassName(typeId, cls.getName());
+ registered = registerUserClassName(typeId, cls.getName(), false);
BinarySerializer serializer = serializerForClass(cls);
private BinaryClassDescriptor registerUserClassDescriptor(BinaryClassDescriptor desc) {
boolean registered;
- registered = registerUserClassName(desc.typeId(), desc.describedClass().getName());
+ registered = registerUserClassName(desc.typeId(), desc.describedClass().getName(), false);
if (registered) {
BinarySerializer serializer = desc.initialSerializer();
*
* @param typeId Type ID.
* @param clsName Class Name.
+ * @param failIfUnregistered If {@code true} then throw {@link UnregisteredBinaryTypeException} with
+ * {@link MappingExchangeResult} future instead of synchronously awaiting for its completion.
* @return {@code True} if the mapping was registered successfully.
*/
- public boolean registerUserClassName(int typeId, String clsName) {
+ public boolean registerUserClassName(int typeId, String clsName, boolean failIfUnregistered) {
IgniteCheckedException e = null;
boolean res = false;
try {
- res = marshCtx.registerClassName(JAVA_ID, typeId, clsName);
+ res = marshCtx.registerClassName(JAVA_ID, typeId, clsName, failIfUnregistered);
}
catch (DuplicateTypeIdException dupEx) {
// Ignore if trying to register mapped type name of the already registered class name and vise versa
try {
write0(obj, writer);
}
+ catch (UnregisteredClassException | UnregisteredBinaryTypeException ex) {
+ throw ex;
+ }
catch (Exception ex) {
- if (ex instanceof UnregisteredClassException || ex instanceof UnregisteredBinaryTypeException)
- throw ex;
-
if (S.INCLUDE_SENSITIVE && !F.isEmpty(name))
throw new BinaryObjectException("Failed to write field [name=" + name + ']', ex);
else
if (affFieldName0 == null)
affFieldName0 = ctx.affinityKeyFieldName(typeId);
- ctx.registerUserClassName(typeId, typeName);
+ ctx.registerUserClassName(typeId, typeName, writer.failIfUnregistered());
ctx.updateMetadata(typeId, new BinaryMetadata(typeId, typeName, fieldsMeta, affFieldName0,
Collections.singleton(curSchema), false, null), writer.failIfUnregistered());
private Map<Integer, String> cache = new ConcurrentHashMap<>();
/** {@inheritDoc} */
- @Override public boolean registerClassName(byte platformId, int typeId, String clsName)
- throws IgniteCheckedException {
+ @Override public boolean registerClassName(
+ byte platformId,
+ int typeId,
+ String clsName,
+ boolean failIfUnregistered
+ ) throws IgniteCheckedException {
if (platformId != MarshallerPlatformIds.JAVA_ID)
throw new IllegalArgumentException("platformId");
}
/** {@inheritDoc} */
+ @Override
+ @Deprecated
+ public boolean registerClassName(byte platformId, int typeId, String clsName) throws IgniteCheckedException {
+ return registerClassName(platformId, typeId, clsName, false);
+ }
+
+ /** {@inheritDoc} */
@Override public boolean registerClassNameLocally(byte platformId, int typeId, String clsName) {
if (platformId != MarshallerPlatformIds.JAVA_ID)
throw new IllegalArgumentException("platformId");
boolean registered;
try {
- registered = ctx.registerClassName(JAVA_ID, typeId, cls.getName());
+ registered = ctx.registerClassName(JAVA_ID, typeId, cls.getName(), false);
}
catch (Exception e) {
throw new IOException("Failed to register class: " + cls.getName(), e);
return null;
}
+ catch (UnregisteredClassException | UnregisteredBinaryTypeException e) {
+ throw e;
+ }
catch (Exception e) {
- if (e instanceof UnregisteredClassException || e instanceof UnregisteredBinaryTypeException)
- throw (IgniteException)e;
-
writeObj = invokeEntry.valObj;
return new IgniteBiTuple<>(null, e);
GridFutureAdapter<MetadataUpdateResult> fut =
transport.awaitMetadataUpdate(typeId, metaHolder.pendingVersion());
+ if (failIfUnregistered && !fut.isDone())
+ throw new UnregisteredBinaryTypeException(typeId, fut);
+
fut.get();
}
+
return;
}
if (failIfUnregistered)
- throw new UnregisteredBinaryTypeException(
- "Attempted to update binary metadata inside a critical synchronization block (will be " +
- "automatically retried). This exception must not be wrapped to any other exception class. " +
- "If you encounter this exception outside of EntryProcessor, please report to Apache Ignite " +
- "dev-list.",
- typeId, mergedMeta);
+ throw new UnregisteredBinaryTypeException(typeId, mergedMeta);
long t0 = System.nanoTime();
.binaryContext().descriptorForClass(ex.cls(), false, false);
}
catch (UnregisteredBinaryTypeException ex) {
+ if (ex.future() != null) {
+ // Wait for the future that couldn't be processed because of
+ // IgniteThread#isForbiddenToRequestBinaryMetadata flag being true. Usually this means
+ // that awaiting for the future right there would lead to potential deadlock if
+ // continuous queries are used in parallel with entry processor.
+ ex.future().get();
+
+ // Retry and don't update current binary metadata, because it most likely already exists.
+ continue;
+ }
+
IgniteCacheObjectProcessor cacheObjProc = ctx.cacheObjects();
assert cacheObjProc instanceof CacheObjectBinaryProcessorImpl;
ctx.validateKeyAndValue(entry.key(), updated);
}
}
+ catch (UnregisteredClassException | UnregisteredBinaryTypeException e) {
+ throw e;
+ }
catch (Exception e) {
- if (e instanceof UnregisteredClassException || e instanceof UnregisteredBinaryTypeException)
- throw (IgniteException) e;
-
curInvokeRes = CacheInvokeResult.fromError(e);
updated = old;
String typeName = reader.readString();
return platformContext().kernalContext().marshallerContext()
- .registerClassName(MarshallerPlatformIds.DOTNET_ID, typeId, typeName)
+ .registerClassName(MarshallerPlatformIds.DOTNET_ID, typeId, typeName, false)
? TRUE : FALSE;
}
}
/** {@inheritDoc} */
@Override public ClientResponse process(ClientConnectionContext ctx) {
try {
- boolean res = ctx.kernalContext().marshallerContext().registerClassName(platformId, typeId, typeName);
+ boolean res = ctx.kernalContext().marshallerContext()
+ .registerClassName(platformId, typeId, typeName, false);
return new ClientBooleanResponse(requestId(), res);
}
package org.apache.ignite.marshaller;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.UnregisteredBinaryTypeException;
import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.marshaller.jdk.JdkMarshaller;
* @param platformId Id of a platform (java, .NET, etc.) to register mapping for.
* @param typeId Type ID.
* @param clsName Class name.
+ * @param failIfUnregistered If {@code true} then throw {@link UnregisteredBinaryTypeException} with
+ * registration future instead of synchronously awaiting for its completion.
* @return {@code True} if mapping was registered successfully.
* @throws IgniteCheckedException In case of error.
*/
- public boolean registerClassName(byte platformId, int typeId, String clsName) throws IgniteCheckedException;
+ public default boolean registerClassName(
+ byte platformId,
+ int typeId,
+ String clsName,
+ boolean failIfUnregistered
+ ) throws IgniteCheckedException {
+ return registerClassName(platformId, typeId, clsName);
+ }
+
+ /**
+ * Same as {@link MarshallerContext#registerClassName(byte, int, java.lang.String, boolean)} but with shortened
+ * parameters list.
+ *
+ * @deprecated Use {@link MarshallerContext#registerClassName(byte, int, java.lang.String, boolean)} instead.
+ * This particular method will be deleted in future releases.
+ */
+ @Deprecated
+ public boolean registerClassName(
+ byte platformId,
+ int typeId,
+ String clsName
+ ) throws IgniteCheckedException;
/**
* Method to register typeId->class name mapping in marshaller context <b>on local node only</b>.
}
/**
+ * @return {@code True} if thread is not allowed to request binary metadata to avoid potential deadlock.
+ */
+ public static boolean currentThreadCanRequestBinaryMetadata() {
+ IgniteThread curThread = current();
+
+ return curThread == null || !curThread.isForbiddenToRequestBinaryMetadata();
+ }
+
+ /**
* Callback before entry processor execution is started.
*/
public static void onEntryProcessorEntered(boolean holdsTopLock) {
@Test
public void testDeclaredBodyEnum() throws Exception {
final MarshallerContextTestImpl ctx = new MarshallerContextTestImpl();
- ctx.registerClassName((byte)0, 1, EnumObject.class.getName());
- ctx.registerClassName((byte)0, 2, DeclaredBodyEnum.class.getName());
+ ctx.registerClassName((byte)0, 1, EnumObject.class.getName(), false);
+ ctx.registerClassName((byte)0, 2, DeclaredBodyEnum.class.getName(), false);
BinaryMarshaller marsh = binaryMarshaller();
marsh.setContext(ctx);
private static class MarshallerContextWithNoStorage implements MarshallerContext {
/** {@inheritDoc} */
@Override public boolean registerClassName(
- byte platformId,
- int typeId,
- String clsName
+ byte platformId,
+ int typeId,
+ String clsName
) throws IgniteCheckedException {
return false;
}
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import javax.cache.Cache;
import javax.cache.processor.EntryProcessor;
import javax.cache.processor.EntryProcessorException;
import javax.cache.processor.MutableEntry;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
+import org.apache.ignite.binary.BinaryObjectBuilder;
+import org.apache.ignite.cache.CacheEntryProcessor;
+import org.apache.ignite.cache.query.ContinuousQuery;
+import org.apache.ignite.cache.query.QueryCursor;
+import org.apache.ignite.cache.query.ScanQuery;
+import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.resources.IgniteInstanceResource;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.junit.After;
import org.junit.Test;
+import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+import static org.apache.ignite.cache.PartitionLossPolicy.READ_WRITE_SAFE;
+
/**
*
*/
.setPeerClassLoadingEnabled(true);
}
+ /** Stop all grids after each test. */
+ @After
+ public void stopAllGridsAfterTest() {
+ stopAllGrids();
+ }
+
/**
* @throws Exception If failed;
*/
}
/**
+ * Continuously execute multiple EntryProcessors with having continuous queries in parallel.
+ * This used to lead to several deadlocks.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testContinuousQueryAndBinaryObjectBuilder() throws Exception {
+ startGrids(3).cluster().active(true);
+
+ grid(0).createCache(new CacheConfiguration<>()
+ .setName(CACHE_NAME)
+ .setAtomicityMode(ATOMIC)
+ .setBackups(2)
+ .setCacheMode(PARTITIONED)
+ .setWriteSynchronizationMode(FULL_SYNC)
+ .setPartitionLossPolicy(READ_WRITE_SAFE)
+ );
+
+ IgniteEx client1 = startGrid(getConfiguration().setIgniteInstanceName("client1").setClientMode(true));
+ IgniteEx client2 = startGrid(getConfiguration().setIgniteInstanceName("client2").setClientMode(true));
+
+ AtomicBoolean stop = new AtomicBoolean();
+ AtomicInteger keyCntr = new AtomicInteger();
+ AtomicInteger binaryTypeCntr = new AtomicInteger();
+
+ /** */
+ class MyEntryProcessor implements CacheEntryProcessor<Object, Object, Object> {
+ /** Cached int value retrieved from {@code binaryTypeCntr} variable. */
+ private int i;
+
+ /** */
+ public MyEntryProcessor(int i) {
+ this.i = i;
+ }
+
+ /** */
+ @IgniteInstanceResource
+ Ignite ignite;
+
+ /** {@inheritDoc} */
+ @Override
+ public Object process(MutableEntry<Object, Object> entry, Object... arguments) throws EntryProcessorException {
+ BinaryObjectBuilder builder = ignite.binary().builder("my_type");
+
+ builder.setField("new_field" + i, i);
+
+ entry.setValue(builder.build());
+
+ return null;
+ }
+ }
+
+ IgniteInternalFuture fut1 = GridTestUtils.runMultiThreadedAsync(() -> {
+ IgniteCache<Object, Object> cache = client1.cache(CACHE_NAME).withKeepBinary();
+
+ while (!stop.get()) {
+ Integer key = keyCntr.getAndIncrement();
+
+ cache.put(key, key);
+
+ cache.invoke(key, new MyEntryProcessor(binaryTypeCntr.get()));
+
+ binaryTypeCntr.incrementAndGet();
+ }
+ }, 8, "writer-thread");
+
+ IgniteInternalFuture fut2 = GridTestUtils.runAsync(() -> {
+ IgniteCache<Object, Object> cache = client2.cache(CACHE_NAME).withKeepBinary();
+
+ while (!stop.get()) {
+ ContinuousQuery<Object, Object> qry = new ContinuousQuery<>();
+
+ qry.setInitialQuery(new ScanQuery<>((key, val) -> true));
+
+ qry.setLocalListener(evts -> {});
+
+ //noinspection EmptyTryBlock
+ try (QueryCursor<Cache.Entry<Object, Object>> cursor = cache.query(qry)) {
+ // No-op.
+ }
+ }
+ });
+
+ doSleep(10_000);
+
+ stop.set(true);
+
+ fut1.get(10, TimeUnit.SECONDS);
+ fut2.get(10, TimeUnit.SECONDS);
+ }
+
+ /**
*
*/
private static class CustomProcessor implements EntryProcessor<Integer,
Marshaller marsh = createStandaloneBinaryMarshaller();
marsh.setContext(new MarshallerContext() {
- @Override public boolean registerClassName(byte platformId, int typeId, String clsName) {
+ @Override public boolean registerClassName(
+ byte platformId,
+ int typeId,
+ String clsName
+ ) {
return true;
}
}
/** {@inheritDoc} */
- @Override public boolean registerClassName(byte platformId, int typeId, String clsName) throws IgniteCheckedException {
+ @Override public boolean registerClassName(byte platformId, int typeId, String clsName, boolean failIfUnregistered) throws IgniteCheckedException {
if (excluded != null && excluded.contains(clsName))
return false;
projects / ignite.git / commitdiff
