You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ni...@apache.org on 2022/04/18 17:13:12 UTC
[ignite] branch master updated: IGNITE-14449 CDC binary meta change events introduced
This is an automated email from the ASF dual-hosted git repository.
nizhikov pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new acbf7d52fdc IGNITE-14449 CDC binary meta change events introduced
acbf7d52fdc is described below
commit acbf7d52fdcb964f2eefdebc166b6488f1a18302
Author: Nikolay Izhikov <ni...@apache.org>
AuthorDate: Mon Apr 18 20:12:43 2022 +0300
IGNITE-14449 CDC binary meta change events introduced
---
.../meta/subcommands/MetadataRemoveCommand.java | 3 +-
.../java/org/apache/ignite/cdc/CdcConsumer.java | 34 +++-
.../java/org/apache/ignite/cdc/TypeMapping.java | 51 ++++++
.../internal/MarshallerMappingFileStore.java | 72 ++-------
.../apache/ignite/internal/binary/BinaryUtils.java | 92 +++++++++++
.../ignite/internal/cdc/CdcConsumerState.java | 163 ++++++++++++++++---
.../org/apache/ignite/internal/cdc/CdcMain.java | 173 +++++++++++++++++----
.../ignite/internal/cdc/TypeMappingImpl.java | 68 ++++++++
.../ignite/internal/cdc/WalRecordsConsumer.java | 18 +++
.../cache/binary/BinaryMetadataFileStore.java | 41 +++--
.../binary/CacheObjectBinaryProcessorImpl.java | 20 ++-
.../checkpoint/CheckpointMarkersStorage.java | 3 +-
.../cacheobject/IgniteCacheObjectProcessor.java | 9 ++
.../org/apache/ignite/cdc/AbstractCdcTest.java | 49 +++++-
.../java/org/apache/ignite/cdc/CdcSelfTest.java | 33 +++-
.../ConcurrentMappingFileReadWriteTest.java | 3 +-
...gnitePdsBinaryMetadataOnClusterRestartTest.java | 3 +-
.../org/apache/ignite/internal/cdc/SqlCdcTest.java | 121 +++++++++++++-
.../apache/ignite/cdc/CdcConfigurationTest.java | 11 ++
19 files changed, 832 insertions(+), 135 deletions(-)
diff --git a/modules/control-utility/src/main/java/org/apache/ignite/internal/commandline/meta/subcommands/MetadataRemoveCommand.java b/modules/control-utility/src/main/java/org/apache/ignite/internal/commandline/meta/subcommands/MetadataRemoveCommand.java
index 08e731b89e2..38a8315c4f1 100644
--- a/modules/control-utility/src/main/java/org/apache/ignite/internal/commandline/meta/subcommands/MetadataRemoveCommand.java
+++ b/modules/control-utility/src/main/java/org/apache/ignite/internal/commandline/meta/subcommands/MetadataRemoveCommand.java
@@ -23,6 +23,7 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.util.logging.Logger;
import org.apache.ignite.internal.binary.BinaryMetadata;
+import org.apache.ignite.internal.binary.BinaryUtils;
import org.apache.ignite.internal.commandline.CommandArgIterator;
import org.apache.ignite.internal.commandline.CommandLogger;
import org.apache.ignite.internal.commandline.meta.MetadataCommand;
@@ -92,7 +93,7 @@ public class MetadataRemoveCommand
BinaryMetadata m = res.metadata();
if (outFile == null)
- outFile = FS.getPath(m.typeId() + ".bin");
+ outFile = FS.getPath(BinaryUtils.binaryMetaFileName(m.typeId()));
try (OutputStream os = Files.newOutputStream(outFile)) {
os.write(res.metadataMarshalled());
diff --git a/modules/core/src/main/java/org/apache/ignite/cdc/CdcConsumer.java b/modules/core/src/main/java/org/apache/ignite/cdc/CdcConsumer.java
index 1a13465013b..c95bc5ed12c 100644
--- a/modules/core/src/main/java/org/apache/ignite/cdc/CdcConsumer.java
+++ b/modules/core/src/main/java/org/apache/ignite/cdc/CdcConsumer.java
@@ -18,7 +18,10 @@
package org.apache.ignite.cdc;
import java.util.Iterator;
+import org.apache.ignite.IgniteBinary;
import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.binary.BinaryIdMapper;
+import org.apache.ignite.binary.BinaryType;
import org.apache.ignite.cache.CacheEntryVersion;
import org.apache.ignite.internal.cdc.CdcMain;
import org.apache.ignite.internal.processors.metric.MetricRegistry;
@@ -71,9 +74,38 @@ public interface CdcConsumer {
*/
public boolean onEvents(Iterator<CdcEvent> events);
+ /**
+ * Handles new binary types. State of the types processing will be stored after method invocation
+ * and ongoing notifications after CDC application fail/restart will be continued for newly created/updates types.
+ * Invoked before {@link #onEvents(Iterator)}.
+ *
+ * Note, unlike {@link #onEvents(Iterator)} this method MUST process all types or CDC will fail.
+ * Because, in time of invocation {@link #onEvents(Iterator)} all changed types must be available on destionation.
+ *
+ * @param types Binary types iterator.
+ * @see IgniteBinary
+ * @see IgniteBinary#type(int)
+ * @see IgniteBinary#type(Class)
+ * @see IgniteBinary#type(String)
+ * @see IgniteBinary#types()
+ */
+ public void onTypes(Iterator<BinaryType> types);
+
+ /**
+ * Handles new mappings from type name to id. State of the types processing will be stored after method invocation
+ * and ongoing notifications after CDC application fail/restart will be continued for newly created/updates mappings.
+ * Invoked before both {@link #onEvents(Iterator)} and {@link #onTypes(Iterator)}.
+ *
+ * @param mappings Binary mapping iterator.
+ * @see IgniteBinary
+ * @see IgniteBinary#typeId(String)
+ * @see BinaryIdMapper
+ */
+ public void onMappings(Iterator<TypeMapping> mappings);
+
/**
* Stops the consumer.
- * This methods can be invoked only after {@link #start(MetricRegistry)}.
+ * This method can be invoked only after {@link #start(MetricRegistry)}.
*/
public void stop();
}
diff --git a/modules/core/src/main/java/org/apache/ignite/cdc/TypeMapping.java b/modules/core/src/main/java/org/apache/ignite/cdc/TypeMapping.java
new file mode 100644
index 00000000000..cda4d73ac3a
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/cdc/TypeMapping.java
@@ -0,0 +1,51 @@
+/*
+ * 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.cdc;
+
+import java.io.Serializable;
+import java.util.Iterator;
+import org.apache.ignite.IgniteBinary;
+import org.apache.ignite.binary.BinaryIdMapper;
+import org.apache.ignite.binary.BinaryObject;
+import org.apache.ignite.lang.IgniteExperimental;
+import org.apache.ignite.marshaller.MarshallerContext;
+import org.apache.ignite.platform.PlatformType;
+
+/**
+ * Ignite maps type name to some id.
+ * Mapping data required to perform deserialization of {@link BinaryObject}.
+ * Note, {@link #typeName()} can represent not only Java class names.
+ * For other supported platforms take a look at {@link PlatformType} values.
+ *
+ * @see BinaryObject
+ * @see IgniteBinary#typeId(String)
+ * @see BinaryIdMapper
+ * @see CdcConsumer#onMappings(Iterator)
+ * @see MarshallerContext#registerClassName(byte, int, String, boolean)
+ */
+@IgniteExperimental
+public interface TypeMapping extends Serializable {
+ /** @return Type id. */
+ public int typeId();
+
+ /** @return Type name. */
+ public String typeName();
+
+ /** @return Platform type. */
+ public PlatformType platformType();
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/MarshallerMappingFileStore.java b/modules/core/src/main/java/org/apache/ignite/internal/MarshallerMappingFileStore.java
index 960de351182..c719710bde9 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/MarshallerMappingFileStore.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/MarshallerMappingFileStore.java
@@ -17,12 +17,9 @@
package org.apache.ignite.internal;
-import java.io.BufferedReader;
import java.io.File;
-import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
-import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.Writer;
import java.nio.charset.StandardCharsets;
@@ -31,6 +28,7 @@ import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.locks.Lock;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.internal.binary.BinaryUtils;
import org.apache.ignite.internal.util.GridStripedLock;
import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.internal.util.typedef.internal.U;
@@ -38,6 +36,8 @@ import org.apache.ignite.marshaller.MarshallerContext;
import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+import static org.apache.ignite.internal.binary.BinaryUtils.MAPPING_FILE_EXTENSION;
+import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.TMP_SUFFIX;
/**
* File-based persistence provider for {@link MarshallerContextImpl}.
@@ -48,9 +48,6 @@ import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
* when a classname is requested but is not presented in local cache of {@link MarshallerContextImpl}.
*/
final class MarshallerMappingFileStore {
- /** */
- private static final String FILE_EXTENSION = ".classname";
-
/** */
private static final GridStripedLock fileLock = new GridStripedLock(32);
@@ -85,9 +82,9 @@ final class MarshallerMappingFileStore {
* @param typeName Type name.
*/
public void writeMapping(byte platformId, int typeId, String typeName) {
- String fileName = getFileName(platformId, typeId);
+ String fileName = BinaryUtils.mappingFileName(platformId, typeId);
- File tmpFile = new File(mappingDir, fileName + ThreadLocalRandom.current().nextInt() + ".tmp");
+ File tmpFile = new File(mappingDir, fileName + ThreadLocalRandom.current().nextInt() + TMP_SUFFIX);
File file = new File(mappingDir, fileName);
Lock lock = fileLock(fileName);
@@ -119,7 +116,7 @@ final class MarshallerMappingFileStore {
* @param typeId Type id.
*/
public String readMapping(byte platformId, int typeId) {
- return readMapping(getFileName(platformId, typeId));
+ return readMapping(BinaryUtils.mappingFileName(platformId, typeId));
}
/**
@@ -131,20 +128,7 @@ final class MarshallerMappingFileStore {
lock.lock();
try {
- File file = new File(mappingDir, fileName);
-
- try (FileInputStream in = new FileInputStream(file)) {
- try (BufferedReader reader = new BufferedReader(new InputStreamReader(in, StandardCharsets.UTF_8))) {
- if (file.length() == 0)
- return null;
-
- return reader.readLine();
- }
-
- }
- catch (IOException ignored) {
- return null;
- }
+ return BinaryUtils.readMapping(new File(mappingDir, fileName));
}
finally {
lock.unlock();
@@ -158,10 +142,15 @@ final class MarshallerMappingFileStore {
* @param marshCtx Marshaller context to register mappings.
*/
void restoreMappings(MarshallerContext marshCtx) throws IgniteCheckedException {
- for (File file : mappingDir.listFiles()) {
+ File[] files = mappingDir.listFiles(BinaryUtils::notTmpFile);
+
+ if (files == null)
+ return;
+
+ for (File file : files) {
String name = file.getName();
- byte platformId = getPlatformId(name);
+ byte platformId = BinaryUtils.mappedFilePlatformId(name);
int typeId = getTypeId(name);
@@ -213,7 +202,7 @@ final class MarshallerMappingFileStore {
"marshaller"
);
- File legacyTmpDir = new File(legacyDir.toString() + ".tmp");
+ File legacyTmpDir = new File(legacyDir + TMP_SUFFIX);
if (legacyTmpDir.exists() && !IgniteUtils.delete(legacyTmpDir))
throw new IgniteCheckedException("Failed to delete legacy marshaller mappings dir: "
@@ -241,27 +230,6 @@ final class MarshallerMappingFileStore {
}
}
- /**
- * @param fileName Name of file with marshaller mapping information.
- * @throws IgniteCheckedException If file name format is broken.
- */
- private byte getPlatformId(String fileName) throws IgniteCheckedException {
- String lastSymbol = fileName.substring(fileName.length() - 1);
-
- byte platformId;
-
- try {
- platformId = Byte.parseByte(lastSymbol);
- }
- catch (NumberFormatException e) {
- throw new IgniteCheckedException("Reading marshaller mapping from file "
- + fileName
- + " failed; last symbol of file name is expected to be numeric.", e);
- }
-
- return platformId;
- }
-
/**
* @param fileName Name of file with marshaller mapping information.
* @throws IgniteCheckedException If file name format is broken.
@@ -270,7 +238,7 @@ final class MarshallerMappingFileStore {
int typeId;
try {
- typeId = Integer.parseInt(fileName.substring(0, fileName.indexOf(FILE_EXTENSION)));
+ typeId = Integer.parseInt(fileName.substring(0, fileName.indexOf(MAPPING_FILE_EXTENSION)));
}
catch (NumberFormatException e) {
throw new IgniteCheckedException("Reading marshaller mapping from file "
@@ -281,14 +249,6 @@ final class MarshallerMappingFileStore {
return typeId;
}
- /**
- * @param platformId Platform id.
- * @param typeId Type id.
- */
- String getFileName(byte platformId, int typeId) {
- return typeId + FILE_EXTENSION + platformId;
- }
-
/**
* @param fileName File name.
* @return Lock instance.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryUtils.java
index dafb8254790..f3203a627e1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryUtils.java
@@ -17,8 +17,13 @@
package org.apache.ignite.internal.binary;
+import java.io.BufferedReader;
import java.io.ByteArrayInputStream;
import java.io.Externalizable;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStreamReader;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.Array;
@@ -28,6 +33,7 @@ import java.lang.reflect.Modifier;
import java.lang.reflect.Proxy;
import java.math.BigDecimal;
import java.math.BigInteger;
+import java.nio.charset.StandardCharsets;
import java.sql.Time;
import java.sql.Timestamp;
import java.util.ArrayList;
@@ -48,6 +54,7 @@ import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListSet;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.binary.BinaryCollectionFactory;
import org.apache.ignite.binary.BinaryInvalidTypeException;
@@ -71,15 +78,26 @@ import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.platform.PlatformType;
import org.jetbrains.annotations.Nullable;
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_BINARY_MARSHALLER_USE_STRING_SERIALIZATION_VER_2;
+import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.TMP_SUFFIX;
/**
* Binary utils.
*/
public class BinaryUtils {
+ /** Binary metadata file suffix. */
+ public static final String METADATA_FILE_SUFFIX = ".bin";
+
+ /**
+ * Actual file name "{type_id}.classname{platform_id}".
+ * Where {@code type_id} is integer type id and {@code platform_id} is byte from {@link PlatformType}
+ */
+ public static final String MAPPING_FILE_EXTENSION = ".classname";
+
/** */
public static final Map<Class<?>, Byte> PLAIN_CLASS_TO_FLAG = new HashMap<>();
@@ -2526,6 +2544,80 @@ public class BinaryUtils {
return ctx.metadata(obj.typeId());
}
+ /**
+ * @param typeId Type id.
+ * @return Binary metadata file name.
+ */
+ public static String binaryMetaFileName(int typeId) {
+ return typeId + METADATA_FILE_SUFFIX;
+ }
+
+ /** @param fileName Name of file with marshaller mapping information. */
+ public static int mappedTypeId(String fileName) {
+ try {
+ return Integer.parseInt(fileName.substring(0, fileName.indexOf(MAPPING_FILE_EXTENSION)));
+ }
+ catch (NumberFormatException e) {
+ throw new IgniteException("Reading marshaller mapping from file "
+ + fileName
+ + " failed; type ID is expected to be numeric.", e);
+ }
+ }
+
+ /** @param fileName Name of file with marshaller mapping information. */
+ public static byte mappedFilePlatformId(String fileName) {
+ try {
+ return Byte.parseByte(fileName.substring(fileName.length() - 1));
+ }
+ catch (NumberFormatException e) {
+ throw new IgniteException("Reading marshaller mapping from file "
+ + fileName
+ + " failed; last symbol of file name is expected to be numeric.", e);
+ }
+ }
+
+ /** @param file File. */
+ public static String readMapping(File file) {
+ try (FileInputStream in = new FileInputStream(file)) {
+ try (BufferedReader reader = new BufferedReader(new InputStreamReader(in, StandardCharsets.UTF_8))) {
+ if (file.length() == 0)
+ return null;
+
+ return reader.readLine();
+ }
+
+ }
+ catch (IOException ignored) {
+ return null;
+ }
+ }
+
+ /**
+ * @param platformId Platform id.
+ * @param typeId Type id.
+ */
+ public static String mappingFileName(byte platformId, int typeId) {
+ return typeId + MAPPING_FILE_EXTENSION + platformId;
+ }
+
+ /**
+ * @param f File.
+ * @return {@code True} if file is regular(not temporary).
+ */
+ public static boolean notTmpFile(File f) {
+ return !f.getName().endsWith(TMP_SUFFIX);
+ }
+
+ /**
+ * @param fileName File name.
+ * @return Type id
+ * @see #binaryMetaFileName(int)
+ * @see #METADATA_FILE_SUFFIX
+ */
+ public static int typeId(String fileName) {
+ return Integer.parseInt(fileName.substring(0, fileName.length() - METADATA_FILE_SUFFIX.length()));
+ }
+
/**
* Get predefined write-replacer associated with class.
*
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcConsumerState.java b/modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcConsumerState.java
index a9ec369560f..e00391cab71 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcConsumerState.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcConsumerState.java
@@ -18,13 +18,22 @@
package org.apache.ignite.internal.cdc;
import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutput;
+import java.io.ObjectOutputStream;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
+import java.util.HashMap;
+import java.util.HashSet;
import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Supplier;
import org.apache.ignite.IgniteException;
+import org.apache.ignite.IgniteLogger;
import org.apache.ignite.cdc.CdcConsumer;
import org.apache.ignite.cdc.CdcEvent;
import org.apache.ignite.internal.pagemem.wal.record.DataEntry;
@@ -51,28 +60,54 @@ import static org.apache.ignite.internal.processors.cache.persistence.wal.WALPoi
*/
public class CdcConsumerState {
/** */
- public static final String STATE_FILE_NAME = "cdc-state" + FILE_SUFFIX;
+ public static final String WAL_STATE_FILE_NAME = "cdc-wal-state" + FILE_SUFFIX;
- /** State file. */
- private final Path state;
+ /** */
+ public static final String TYPES_STATE_FILE_NAME = "cdc-types-state" + FILE_SUFFIX;
+
+ /** */
+ public static final String MAPPINGS_STATE_FILE_NAME = "cdc-mappings-state" + FILE_SUFFIX;
+
+ /** Log. */
+ private final IgniteLogger log;
+
+ /** WAL pointer state file. */
+ private final Path walPtr;
+
+ /** Temp WAL pointer state file. */
+ private final Path tmpWalPtr;
+
+ /** Types state file. */
+ private final Path types;
- /** Temp state file. */
- private final Path tmp;
+ /** Temp types state file. */
+ private final Path tmpTypes;
+
+ /** Mappings state file. */
+ private final Path mappings;
+
+ /** Mappings types state file. */
+ private final Path tmpMappings;
/**
* @param stateDir State directory.
*/
- public CdcConsumerState(Path stateDir) {
- state = stateDir.resolve(STATE_FILE_NAME);
- tmp = stateDir.resolve(STATE_FILE_NAME + TMP_SUFFIX);
+ public CdcConsumerState(IgniteLogger log, Path stateDir) {
+ this.log = log.getLogger(CdcConsumerState.class);
+ walPtr = stateDir.resolve(WAL_STATE_FILE_NAME);
+ tmpWalPtr = stateDir.resolve(WAL_STATE_FILE_NAME + TMP_SUFFIX);
+ types = stateDir.resolve(TYPES_STATE_FILE_NAME);
+ tmpTypes = stateDir.resolve(TYPES_STATE_FILE_NAME + TMP_SUFFIX);
+ mappings = stateDir.resolve(MAPPINGS_STATE_FILE_NAME);
+ tmpMappings = stateDir.resolve(MAPPINGS_STATE_FILE_NAME + TMP_SUFFIX);
}
/**
- * Saves state to file.
+ * Saves WAL consumption state to file.
*
* @param state WAL pointer and index of {@link DataEntry} inside {@link DataRecord}.
*/
- public void save(T2<WALPointer, Integer> state) throws IOException {
+ public void saveWal(T2<WALPointer, Integer> state) throws IOException {
ByteBuffer buf = ByteBuffer.allocate(POINTER_SIZE);
buf.putLong(state.get1().index());
@@ -80,28 +115,51 @@ public class CdcConsumerState {
buf.putInt(state.get2());
buf.flip();
- try (FileChannel ch = FileChannel.open(tmp, StandardOpenOption.CREATE, StandardOpenOption.WRITE)) {
+ try (FileChannel ch = FileChannel.open(tmpWalPtr, StandardOpenOption.CREATE, StandardOpenOption.WRITE)) {
ch.write(buf);
ch.force(true);
}
- Files.move(tmp, this.state, ATOMIC_MOVE, REPLACE_EXISTING);
+ Files.move(tmpWalPtr, walPtr, ATOMIC_MOVE, REPLACE_EXISTING);
+ }
+
+ /**
+ * Saves binary types state to file.
+ *
+ * @param typesState State of types.
+ */
+ public void saveTypes(Map<Integer, Long> typesState) throws IOException {
+ save(typesState, tmpTypes, types);
+ }
+
+ /**
+ * Saves types mappings state to file.
+ *
+ * @param mappingsState Mappings state.
+ */
+ public void saveMappings(Set<T2<Integer, Byte>> mappingsState) throws IOException {
+ save(mappingsState, tmpMappings, mappings);
}
/**
- * Loads CDC state from file.
+ * Loads CDC consumer state from file.
*
* @return Saved state.
*/
- public T2<WALPointer, Integer> load() {
- if (!Files.exists(state))
+ public T2<WALPointer, Integer> loadWalState() {
+ if (!Files.exists(walPtr))
return null;
- try (FileChannel ch = FileChannel.open(state, StandardOpenOption.READ)) {
+ T2<WALPointer, Integer> state;
+
+ try (FileChannel ch = FileChannel.open(walPtr, StandardOpenOption.READ)) {
ByteBuffer buf = ByteBuffer.allocate(POINTER_SIZE);
- ch.read(buf);
+ int read = ch.read(buf);
+
+ if (read != POINTER_SIZE)
+ return null;
buf.flip();
@@ -109,11 +167,78 @@ public class CdcConsumerState {
int offset = buf.getInt();
int entryIdx = buf.getInt();
- return new T2<>(new WALPointer(idx, offset, 0), entryIdx);
+ state = new T2<>(new WALPointer(idx, offset, 0), entryIdx);
}
catch (IOException e) {
- throw new IgniteException("Failed to read state [file=" + state + ']', e);
+ throw new IgniteException("Failed to read state [file=" + walPtr + ']', e);
+ }
+
+ if (log.isInfoEnabled())
+ log.info("Initial WAL state loaded [ptr=" + state.get1() + ", idx=" + state.get2() + ']');
+
+ return state;
+ }
+
+ /**
+ * Loads types mappings state from file.
+ *
+ * @return Saved state.
+ */
+ public Set<T2<Integer, Byte>> loadMappingsState() {
+ Set<T2<Integer, Byte>> state = load(mappings, HashSet::new);
+
+ assert state != null;
+
+ log.info("Initial mappings state loaded [mappingsCnt=" + state.size() + ']');
+
+ if (log.isDebugEnabled()) {
+ for (T2<Integer, Byte> m : state)
+ log.debug("Mapping [typeId=" + m.get1() + ", platform=" + m.get2() + ']');
+ }
+
+ return state;
+ }
+
+ /**
+ * Loads CDC types state from file.
+ *
+ * @return Saved state.
+ */
+ public Map<Integer, Long> loadTypesState() {
+ Map<Integer, Long> state = load(types, HashMap::new);
+
+ assert state != null;
+
+ log.info("Initial types state loaded [typesCnt=" + state.size() + ']');
+
+ if (log.isDebugEnabled()) {
+ for (Map.Entry<Integer, Long> e : state.entrySet())
+ log.debug("Type [typeId=" + e.getKey() + ", lastModified=" + e.getValue() + ']');
+ }
+
+ return state;
+ }
+
+ /** Save object to file. */
+ private <T> void save(T state, Path tmp, Path file) throws IOException {
+ try (ObjectOutput oos = new ObjectOutputStream(Files.newOutputStream(tmp))) {
+ oos.writeObject(state);
}
+ Files.move(tmp, file, ATOMIC_MOVE, REPLACE_EXISTING);
+ }
+
+ /** Loads data from path. */
+ private <D> D load(Path state, Supplier<D> dflt) {
+ if (!Files.exists(state))
+ return dflt.get();
+
+ try (ObjectInputStream ois = new ObjectInputStream(Files.newInputStream(state))) {
+
+ return (D)ois.readObject();
+ }
+ catch (IOException | ClassNotFoundException e) {
+ throw new RuntimeException(e);
+ }
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcMain.java b/modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcMain.java
index c466834d4c0..bb8f2e2b4af 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcMain.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcMain.java
@@ -23,18 +23,23 @@ import java.io.Serializable;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
+import java.util.Arrays;
import java.util.Comparator;
import java.util.HashSet;
import java.util.Iterator;
+import java.util.Map;
+import java.util.Objects;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Stream;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.binary.BinaryType;
import org.apache.ignite.cdc.CdcConfiguration;
import org.apache.ignite.cdc.CdcConsumer;
import org.apache.ignite.cdc.CdcEvent;
+import org.apache.ignite.cdc.TypeMapping;
import org.apache.ignite.configuration.DataRegionConfiguration;
import org.apache.ignite.configuration.DataStorageConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
@@ -42,6 +47,7 @@ import org.apache.ignite.internal.GridComponent;
import org.apache.ignite.internal.GridLoggerProxy;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.MarshallerContextImpl;
+import org.apache.ignite.internal.binary.BinaryUtils;
import org.apache.ignite.internal.cdc.WalRecordsConsumer.DataEntryIterator;
import org.apache.ignite.internal.processors.cache.binary.CacheObjectBinaryProcessorImpl;
import org.apache.ignite.internal.processors.cache.persistence.filename.PdsFolderResolver;
@@ -51,11 +57,13 @@ import org.apache.ignite.internal.processors.cache.persistence.wal.reader.Ignite
import org.apache.ignite.internal.processors.cache.persistence.wal.reader.StandaloneGridKernalContext;
import org.apache.ignite.internal.processors.metric.MetricRegistry;
import org.apache.ignite.internal.processors.metric.impl.AtomicLongMetric;
+import org.apache.ignite.internal.processors.metric.impl.HistogramMetricImpl;
import org.apache.ignite.internal.processors.resource.GridSpringResourceContext;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.platform.PlatformType;
import org.apache.ignite.startup.cmdline.CdcCommandLineStartup;
import static org.apache.ignite.internal.IgniteKernal.NL;
@@ -63,6 +71,7 @@ import static org.apache.ignite.internal.IgniteKernal.SITE;
import static org.apache.ignite.internal.IgniteVersionUtils.ACK_VER_STR;
import static org.apache.ignite.internal.IgniteVersionUtils.COPYRIGHT;
import static org.apache.ignite.internal.IgnitionEx.initializeDefaultMBeanServer;
+import static org.apache.ignite.internal.binary.BinaryUtils.METADATA_FILE_SUFFIX;
import static org.apache.ignite.internal.pagemem.wal.record.WALRecord.RecordType.DATA_RECORD_V2;
import static org.apache.ignite.internal.processors.cache.persistence.wal.FileWriteAheadLogManager.WAL_SEGMENT_FILE_FILTER;
import static org.apache.ignite.internal.processors.metric.impl.MetricUtils.metricName;
@@ -128,6 +137,9 @@ public class CdcMain implements Runnable {
/** Last segment consumption time. */
public static final String LAST_SEG_CONSUMPTION_TIME = "LastSegmentConsumptionTime";
+ /** Metadata update time. */
+ public static final String META_UPDATE = "MetadataUpdateTime";
+
/** Binary metadata metric name. */
public static final String BINARY_META_DIR = "BinaryMetaDir";
@@ -158,12 +170,12 @@ public class CdcMain implements Runnable {
/** Time of last segment consumption. */
private AtomicLongMetric lastSegmentConsumptionTs;
+ /** Metadata update time. */
+ private HistogramMetricImpl metaUpdate;
+
/** Change Data Capture configuration. */
protected final CdcConfiguration cdcCfg;
- /** WAL iterator factory. */
- private final IgniteWalIteratorFactory factory;
-
/** Events consumer. */
private final WalRecordsConsumer<?, ?> consumer;
@@ -179,11 +191,20 @@ public class CdcMain implements Runnable {
/** Marshaller directory. */
private File marshaller;
+ /** Standalone kernal context. */
+ private StandaloneGridKernalContext kctx;
+
/** Change Data Capture state. */
private CdcConsumerState state;
/** Save state to start from. */
- private T2<WALPointer, Integer> initState;
+ private T2<WALPointer, Integer> walState;
+
+ /** Types state. */
+ private Map<Integer, Long> typesState;
+
+ /** Mappings state. */
+ private Set<T2<Integer, Byte>> mappingsState;
/** Stopped flag. */
private volatile boolean stopped;
@@ -215,8 +236,6 @@ public class CdcMain implements Runnable {
}
consumer = new WalRecordsConsumer<>(cdcCfg.getConsumer(), log);
-
- factory = new IgniteWalIteratorFactory(log);
}
/** Runs Change Data Capture. */
@@ -261,7 +280,7 @@ public class CdcMain implements Runnable {
log.info("Ignite node Marshaller [dir=" + marshaller + ']');
}
- StandaloneGridKernalContext kctx = startStandaloneKernal();
+ kctx = startStandaloneKernal();
initMetrics();
@@ -270,14 +289,13 @@ public class CdcMain implements Runnable {
state = createState(cdcDir.resolve(STATE_DIR));
- initState = state.load();
+ walState = state.loadWalState();
+ typesState = state.loadTypesState();
+ mappingsState = state.loadMappingsState();
- if (initState != null) {
- committedSegmentIdx.value(initState.get1().index());
- committedSegmentOffset.value(initState.get1().fileOffset());
-
- if (log.isInfoEnabled())
- log.info("Initial state loaded [ptr=" + initState.get1() + ", idx=" + initState.get2() + ']');
+ if (walState != null) {
+ committedSegmentIdx.value(walState.get1().index());
+ committedSegmentOffset.value(walState.get1().fileOffset());
}
consumer.start(mreg, kctx.metric().registry(metricName("cdc", "consumer")));
@@ -301,7 +319,7 @@ public class CdcMain implements Runnable {
/** Creates consumer state. */
protected CdcConsumerState createState(Path stateDir) {
- return new CdcConsumerState(stateDir);
+ return new CdcConsumerState(log, stateDir);
}
/**
@@ -345,6 +363,7 @@ public class CdcMain implements Runnable {
committedSegmentOffset = mreg.longMetric(COMMITTED_SEG_OFFSET, "Committed segment offset");
lastSegmentConsumptionTs =
mreg.longMetric(LAST_SEG_CONSUMPTION_TIME, "Last time of consumption of WAL segment");
+ metaUpdate = mreg.histogram(META_UPDATE, new long[] {100, 500, 1000}, "Metadata update time");
}
/**
@@ -420,7 +439,7 @@ public class CdcMain implements Runnable {
if (log.isInfoEnabled())
log.info("Processing WAL segment [segment=" + segment + ']');
- lastSegmentConsumptionTs.value(System.currentTimeMillis());
+ updateMetadata();
IgniteWalIteratorFactory.IteratorParametersBuilder builder =
new IgniteWalIteratorFactory.IteratorParametersBuilder()
@@ -436,18 +455,20 @@ public class CdcMain implements Runnable {
long segmentIdx = segmentIndex(segment);
+ lastSegmentConsumptionTs.value(System.currentTimeMillis());
+
curSegmentIdx.value(segmentIdx);
- if (initState != null) {
- if (segmentIdx > initState.get1().index()) {
+ if (walState != null) {
+ if (segmentIdx > walState.get1().index()) {
throw new IgniteException("Found segment greater then saved state. Some events are missed. Exiting! " +
- "[state=" + initState + ", segment=" + segmentIdx + ']');
+ "[state=" + walState + ", segment=" + segmentIdx + ']');
}
- if (segmentIdx < initState.get1().index()) {
+ if (segmentIdx < walState.get1().index()) {
if (log.isInfoEnabled()) {
log.info("Already processed segment found. Skipping and deleting the file [segment=" +
- segmentIdx + ", state=" + initState.get1().index() + ']');
+ segmentIdx + ", state=" + walState.get1().index() + ']');
}
// WAL segment is a hard link to a segment file in the special Change Data Capture folder.
@@ -462,14 +483,14 @@ public class CdcMain implements Runnable {
}
}
- builder.from(initState.get1());
+ builder.from(walState.get1());
}
- try (DataEntryIterator iter = new DataEntryIterator(factory.iterator(builder))) {
- if (initState != null) {
- iter.init(initState.get2());
+ try (DataEntryIterator iter = new DataEntryIterator(new IgniteWalIteratorFactory(log).iterator(builder))) {
+ if (walState != null) {
+ iter.init(walState.get2());
- initState = null;
+ walState = null;
}
boolean interrupted = Thread.interrupted();
@@ -485,7 +506,7 @@ public class CdcMain implements Runnable {
if (log.isDebugEnabled())
log.debug("Saving state [curState=" + curState + ']');
- state.save(curState);
+ state.saveWal(curState);
committedSegmentIdx.value(curState.get1().index());
committedSegmentOffset.value(curState.get1().fileOffset());
@@ -518,6 +539,104 @@ public class CdcMain implements Runnable {
}
}
+ /** Metadata update. */
+ private void updateMetadata() {
+ long start = System.currentTimeMillis();
+
+ updateMappings();
+
+ updateTypes();
+
+ metaUpdate.value(System.currentTimeMillis() - start);
+ }
+
+ /** Search for new or changed {@link BinaryType} and notifies the consumer. */
+ private void updateTypes() {
+ try {
+ Iterator<BinaryType> changedTypes = Files.list(binaryMeta.toPath())
+ .filter(p -> p.toString().endsWith(METADATA_FILE_SUFFIX))
+ .map(p -> {
+ int typeId = BinaryUtils.typeId(p.getFileName().toString());
+ long lastModified = p.toFile().lastModified();
+
+ // Filter out files already in `typesState` with the same last modify date.
+ if (typesState.containsKey(typeId) && lastModified == typesState.get(typeId))
+ return null;
+
+ typesState.put(typeId, lastModified);
+
+ try {
+ kctx.cacheObjects().cacheMetadataLocally(binaryMeta, typeId);
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteException(e);
+ }
+
+ return kctx.cacheObjects().metadata(typeId);
+ })
+ .filter(Objects::nonNull)
+ .iterator();
+
+ if (!changedTypes.hasNext())
+ return;
+
+ consumer.onTypes(changedTypes);
+
+ if (changedTypes.hasNext())
+ throw new IllegalStateException("Consumer should handle all changed types");
+
+ state.saveTypes(typesState);
+ }
+ catch (IOException e) {
+ throw new IgniteException(e);
+ }
+ }
+
+ /** Search for new or changed {@link TypeMapping} and notifies the consumer. */
+ private void updateMappings() {
+ try {
+ File[] files = marshaller.listFiles(BinaryUtils::notTmpFile);
+
+ if (files == null)
+ return;
+
+ Iterator<TypeMapping> changedMappings = Arrays.stream(files)
+ .map(f -> {
+ String fileName = f.getName();
+
+ int typeId = BinaryUtils.mappedTypeId(fileName);
+ byte platformId = BinaryUtils.mappedFilePlatformId(fileName);
+
+ T2<Integer, Byte> state = new T2<>(typeId, platformId);
+
+ if (mappingsState.contains(state))
+ return null;
+
+ mappingsState.add(state);
+
+ return (TypeMapping)new TypeMappingImpl(
+ typeId,
+ BinaryUtils.readMapping(f),
+ platformId == 0 ? PlatformType.JAVA : PlatformType.DOTNET);
+ })
+ .filter(Objects::nonNull)
+ .iterator();
+
+ if (!changedMappings.hasNext())
+ return;
+
+ consumer.onMappings(changedMappings);
+
+ if (changedMappings.hasNext())
+ throw new IllegalStateException("Consumer should handle all changed mappings");
+
+ state.saveMappings(mappingsState);
+ }
+ catch (IOException e) {
+ throw new IgniteException(e);
+ }
+ }
+
/**
* Try locks Change Data Capture directory.
*
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cdc/TypeMappingImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/cdc/TypeMappingImpl.java
new file mode 100644
index 00000000000..e02a7a417ff
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/cdc/TypeMappingImpl.java
@@ -0,0 +1,68 @@
+/*
+ * 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.cdc;
+
+import org.apache.ignite.cdc.TypeMapping;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.platform.PlatformType;
+
+/** */
+public class TypeMappingImpl implements TypeMapping {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ private final int typeId;
+
+ /** */
+ private final String typeName;
+
+ /** */
+ private final PlatformType platform;
+
+ /**
+ * @param typeId Type id.
+ * @param typeName Type name.
+ * @param platform Platform.
+ */
+ public TypeMappingImpl(int typeId, String typeName, PlatformType platform) {
+ this.typeId = typeId;
+ this.typeName = typeName;
+ this.platform = platform;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int typeId() {
+ return typeId;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String typeName() {
+ return typeName;
+ }
+
+ /** {@inheritDoc} */
+ @Override public PlatformType platformType() {
+ return platform;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(TypeMappingImpl.class, this);
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cdc/WalRecordsConsumer.java b/modules/core/src/main/java/org/apache/ignite/internal/cdc/WalRecordsConsumer.java
index 8f87d558b10..7a3c4a1b517 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/cdc/WalRecordsConsumer.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/cdc/WalRecordsConsumer.java
@@ -23,8 +23,10 @@ import java.util.NoSuchElementException;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.binary.BinaryType;
import org.apache.ignite.cdc.CdcConsumer;
import org.apache.ignite.cdc.CdcEvent;
+import org.apache.ignite.cdc.TypeMapping;
import org.apache.ignite.internal.pagemem.wal.WALIterator;
import org.apache.ignite.internal.pagemem.wal.record.DataEntry;
import org.apache.ignite.internal.pagemem.wal.record.DataRecord;
@@ -137,6 +139,22 @@ public class WalRecordsConsumer<K, V> {
return consumer.onEvents(evts);
}
+ /**
+ * Handles new binary types.
+ * @param types Binary types iterator.
+ */
+ public void onTypes(Iterator<BinaryType> types) {
+ consumer.onTypes(types);
+ }
+
+ /**
+ * Handles new mappings.
+ * @param mappings Mappings iterator.
+ */
+ public void onMappings(Iterator<TypeMapping> mappings) {
+ consumer.onMappings(mappings);
+ }
+
/**
* Starts the consumer.
*
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataFileStore.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataFileStore.java
index 18eb9adf335..c412a5dca2b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataFileStore.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataFileStore.java
@@ -46,6 +46,8 @@ import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.internal.util.worker.GridWorker;
import org.apache.ignite.thread.IgniteThread;
+import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.TMP_SUFFIX;
+
/**
* Class handles saving/restoring binary metadata to/from disk.
*
@@ -142,7 +144,7 @@ class BinaryMetadataFileStore {
return;
try {
- File file = new File(metadataDir, binMeta.typeId() + ".bin");
+ File file = new File(metadataDir, BinaryUtils.binaryMetaFileName(binMeta.typeId()));
byte[] marshalled = U.marshal(ctx, binMeta);
@@ -177,7 +179,7 @@ class BinaryMetadataFileStore {
if (!enabled)
return;
- File file = new File(metadataDir, typeId + ".bin");
+ File file = new File(metadataDir, BinaryUtils.binaryMetaFileName(typeId));
if (!file.delete()) {
final String msg = "Failed to remove metadata for typeId: " + typeId;
@@ -201,16 +203,29 @@ class BinaryMetadataFileStore {
if (!enabled)
return;
- for (File file : metadataDir.listFiles()) {
- try (FileInputStream in = new FileInputStream(file)) {
- BinaryMetadata meta = U.unmarshal(ctx.config().getMarshaller(), in, U.resolveClassLoader(ctx.config()));
+ for (File file : metadataDir.listFiles())
+ restoreMetadata(file);
+ }
- metadataLocCache.put(meta.typeId(), new BinaryMetadataHolder(meta, 0, 0));
- }
- catch (Exception e) {
- U.warn(log, "Failed to restore metadata from file: " + file.getName() +
- "; exception was thrown: " + e.getMessage());
- }
+ /**
+ * Restores single type metadata.
+ *
+ * @param typeId Type identifier.
+ */
+ void restoreMetadata(int typeId) {
+ restoreMetadata(new File(metadataDir, BinaryUtils.binaryMetaFileName(typeId)));
+ }
+
+ /** */
+ private void restoreMetadata(File file) {
+ try (FileInputStream in = new FileInputStream(file)) {
+ BinaryMetadata meta = U.unmarshal(ctx.config().getMarshaller(), in, U.resolveClassLoader(ctx.config()));
+
+ metadataLocCache.put(meta.typeId(), new BinaryMetadataHolder(meta, 0, 0));
+ }
+ catch (Exception e) {
+ U.warn(log, "Failed to restore metadata from file: " + file.getName() +
+ "; exception was thrown: " + e.getMessage());
}
}
@@ -238,7 +253,7 @@ class BinaryMetadataFileStore {
* @param typeId typeId of BinaryMetadata to be read.
*/
private BinaryMetadata readMetadata(int typeId) {
- File file = new File(metadataDir, Integer.toString(typeId) + ".bin");
+ File file = new File(metadataDir, BinaryUtils.binaryMetaFileName(typeId));
if (!file.exists())
return null;
@@ -325,7 +340,7 @@ class BinaryMetadataFileStore {
"binary_meta"
), consistendId);
- File legacyTmpDir = new File(legacyDir.toString() + ".tmp");
+ File legacyTmpDir = new File(legacyDir.toString() + TMP_SUFFIX);
if (legacyTmpDir.exists() && !IgniteUtils.delete(legacyTmpDir))
throw new IgniteCheckedException("Failed to delete legacy binary metadata dir: "
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java
index e636d4f854d..4686c8273ce 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java
@@ -530,7 +530,6 @@ public class CacheObjectBinaryProcessorImpl extends GridProcessorAdapter impleme
pArr
);
}
-
}
if (obj instanceof IgniteBiTuple) {
@@ -710,6 +709,11 @@ public class CacheObjectBinaryProcessorImpl extends GridProcessorAdapter impleme
/** {@inheritDoc} */
@Override public void addMetaLocally(int typeId, BinaryType newMeta) throws BinaryObjectException {
+ addMetaLocally(typeId, newMeta, true);
+ }
+
+ /** */
+ private void addMetaLocally(int typeId, BinaryType newMeta, boolean writeToFile) {
assert newMeta != null;
assert newMeta instanceof BinaryTypeImpl;
@@ -722,7 +726,7 @@ public class CacheObjectBinaryProcessorImpl extends GridProcessorAdapter impleme
try {
BinaryMetadata mergedMeta = mergeMetadata(oldMeta, newMeta0);
- if (!ctx.clientNode())
+ if (!ctx.clientNode() && writeToFile)
metadataFileStore.mergeAndWriteMetadata(mergedMeta);
metadataLocCache.put(typeId, new BinaryMetadataHolder(mergedMeta, 0, 0));
@@ -1061,6 +1065,18 @@ public class CacheObjectBinaryProcessorImpl extends GridProcessorAdapter impleme
}
}
+ /** {@inheritDoc} */
+ @Override public void cacheMetadataLocally(File metadataDir, int typeId) throws IgniteCheckedException {
+ if (!metadataDir.exists())
+ return;
+
+ ConcurrentMap<Integer, BinaryMetadataHolder> metaCache = new ConcurrentHashMap<>();
+
+ new BinaryMetadataFileStore(metaCache, ctx, log, metadataDir).restoreMetadata(typeId);
+
+ addMetaLocally(typeId, metaCache.get(typeId).metadata().wrap(binaryContext()), false);
+ }
+
/** {@inheritDoc} */
@Override public BinaryObject buildEnum(String typeName, int ord) throws BinaryObjectException {
A.notNullOrEmpty(typeName, "enum type name");
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/checkpoint/CheckpointMarkersStorage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/checkpoint/CheckpointMarkersStorage.java
index 62a806835ce..969c8c34e18 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/checkpoint/CheckpointMarkersStorage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/checkpoint/CheckpointMarkersStorage.java
@@ -78,7 +78,8 @@ public class CheckpointMarkersStorage {
public static final String EARLIEST_CP_SNAPSHOT_FILE = "cpMapSnapshot.bin";
/** Earliest checkpoint map snapshot temporary file name. */
- private static final String EARLIEST_CP_SNAPSHOT_TMP_FILE = EARLIEST_CP_SNAPSHOT_FILE + ".tmp";
+ private static final String EARLIEST_CP_SNAPSHOT_TMP_FILE =
+ EARLIEST_CP_SNAPSHOT_FILE + FilePageStoreManager.TMP_SUFFIX;
/** Checkpoint map snapshot executor. */
private final Executor checkpointMapSnapshotExecutor;
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessor.java
index 97dc6ec3e9e..68953161430 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessor.java
@@ -317,6 +317,15 @@ public interface IgniteCacheObjectProcessor extends GridProcessor {
*/
public void updateMetadata(File metadataDir, BooleanSupplier stopChecker) throws IgniteCheckedException;
+ /**
+ * Merge the binary metadata file stored in the specified directory.
+ *
+ * @param metadataDir Directory containing binary metadata files.
+ * @param typeId Type id to update.
+ * @throws IgniteCheckedException If failed.
+ */
+ public void cacheMetadataLocally(File metadataDir, int typeId) throws IgniteCheckedException;
+
/**
* @param typeName Type name.
* @param ord ordinal.
diff --git a/modules/core/src/test/java/org/apache/ignite/cdc/AbstractCdcTest.java b/modules/core/src/test/java/org/apache/ignite/cdc/AbstractCdcTest.java
index 3dbe4ff3a18..10d4a8c5176 100644
--- a/modules/core/src/test/java/org/apache/ignite/cdc/AbstractCdcTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/cdc/AbstractCdcTest.java
@@ -32,6 +32,8 @@ import java.util.function.Function;
import javax.management.DynamicMBean;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteException;
+import org.apache.ignite.binary.BinaryBasicIdMapper;
+import org.apache.ignite.binary.BinaryType;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
@@ -117,9 +119,9 @@ public abstract class AbstractCdcTest extends GridCommonAbstractTest {
return new CdcMain(cfg, null, cdcCfg) {
@Override protected CdcConsumerState createState(Path stateDir) {
- return new CdcConsumerState(stateDir) {
- @Override public void save(T2<WALPointer, Integer> state) throws IOException {
- super.save(state);
+ return new CdcConsumerState(log, stateDir) {
+ @Override public void saveWal(T2<WALPointer, Integer> state) throws IOException {
+ super.saveWal(state);
if (!F.isEmpty(conditions)) {
for (GridAbsPredicate p : conditions) {
@@ -306,6 +308,11 @@ public abstract class AbstractCdcTest extends GridCommonAbstractTest {
return commit();
}
+ /** {@inheritDoc} */
+ @Override public void onTypes(Iterator<BinaryType> types) {
+ types.forEachRemaining(t -> assertNotNull(t));
+ }
+
/** */
public abstract void checkEvent(CdcEvent evt);
@@ -330,8 +337,12 @@ public abstract class AbstractCdcTest extends GridCommonAbstractTest {
/** */
public static class UserCdcConsumer extends TestCdcConsumer<Integer> {
+ /** */
+ protected boolean userTypeFound;
+
/** {@inheritDoc} */
@Override public void checkEvent(CdcEvent evt) {
+ assertTrue(userTypeFound);
assertNull(evt.version().otherClusterVersion());
if (evt.value() == null)
@@ -347,6 +358,38 @@ public abstract class AbstractCdcTest extends GridCommonAbstractTest {
@Override public Integer extract(CdcEvent evt) {
return (Integer)evt.key();
}
+
+ /** {@inheritDoc} */
+ @Override public void onTypes(Iterator<BinaryType> types) {
+ types.forEachRemaining(t -> {
+ if (t.typeName().equals(User.class.getName())) {
+ userTypeFound = true;
+
+ assertNotNull(t.field("name"));
+ assertEquals(String.class.getSimpleName(), t.fieldTypeName("name"));
+ assertNotNull(t.field("age"));
+ assertEquals(int.class.getName(), t.fieldTypeName("age"));
+ assertNotNull(t.field("payload"));
+ assertEquals(byte[].class.getSimpleName(), t.fieldTypeName("payload"));
+ }
+
+ assertNotNull(t);
+ });
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onMappings(Iterator<TypeMapping> mappings) {
+ BinaryBasicIdMapper mapper = new BinaryBasicIdMapper();
+
+ mappings.forEachRemaining(m -> {
+ assertNotNull(m);
+
+ String typeName = m.typeName();
+
+ assertFalse(typeName.isEmpty());
+ assertEquals(mapper.typeId(typeName), m.typeId());
+ });
+ }
}
/** */
diff --git a/modules/core/src/test/java/org/apache/ignite/cdc/CdcSelfTest.java b/modules/core/src/test/java/org/apache/ignite/cdc/CdcSelfTest.java
index 2f3bcbe5ce4..a5eed340c5e 100644
--- a/modules/core/src/test/java/org/apache/ignite/cdc/CdcSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/cdc/CdcSelfTest.java
@@ -37,6 +37,7 @@ import java.util.stream.Collectors;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.binary.BinaryType;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.DataRegionConfiguration;
import org.apache.ignite.configuration.DataStorageConfiguration;
@@ -275,6 +276,14 @@ public class CdcSelfTest extends AbstractCdcTest {
return true;
}
+ @Override public void onTypes(Iterator<BinaryType> types) {
+ types.forEachRemaining(t -> assertNotNull(t));
+ }
+
+ @Override public void onMappings(Iterator<TypeMapping> mappings) {
+ mappings.forEachRemaining(m -> assertNotNull(m));
+ }
+
@Override public void stop() {
// No-op.
}
@@ -350,6 +359,14 @@ public class CdcSelfTest extends AbstractCdcTest {
return true;
}
+ @Override public void onTypes(Iterator<BinaryType> types) {
+ types.forEachRemaining(t -> assertNotNull(t));
+ }
+
+ @Override public void onMappings(Iterator<TypeMapping> mappings) {
+ mappings.forEachRemaining(m -> assertNotNull(m));
+ }
+
@Override public void stop() {
// No-op.
}
@@ -570,13 +587,13 @@ public class CdcSelfTest extends AbstractCdcTest {
addData(cache, KEYS_CNT / 2, KEYS_CNT);
- for (int i = 0; i < 3; i++) {
- UserCdcConsumer cnsmr = new UserCdcConsumer() {
- @Override protected boolean commit() {
- return false;
- }
- };
+ UserCdcConsumer cnsmr = new UserCdcConsumer() {
+ @Override protected boolean commit() {
+ return false;
+ }
+ };
+ for (int i = 0; i < 3; i++) {
CdcMain cdc = createCdc(cnsmr, getConfiguration(ign.name()));
IgniteInternalFuture<?> fut = runAsync(cdc);
@@ -588,6 +605,8 @@ public class CdcSelfTest extends AbstractCdcTest {
fut.cancel();
assertTrue(cnsmr.stopped());
+
+ cnsmr.data.clear();
}
AtomicBoolean consumeHalf = new AtomicBoolean(true);
@@ -595,7 +614,7 @@ public class CdcSelfTest extends AbstractCdcTest {
int half = KEYS_CNT / 2;
- UserCdcConsumer cnsmr = new UserCdcConsumer() {
+ cnsmr = new UserCdcConsumer() {
@Override public boolean onEvents(Iterator<CdcEvent> evts) {
if (consumeHalf.get() && F.size(data(UPDATE, cacheId(DEFAULT_CACHE_NAME))) == half) {
// This means that state committed as a result of the previous call.
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/ConcurrentMappingFileReadWriteTest.java b/modules/core/src/test/java/org/apache/ignite/internal/ConcurrentMappingFileReadWriteTest.java
index f34b56950fd..3fe71cfcc33 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/ConcurrentMappingFileReadWriteTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/ConcurrentMappingFileReadWriteTest.java
@@ -26,6 +26,7 @@ import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeoutException;
import org.apache.ignite.binary.BinaryBasicIdMapper;
+import org.apache.ignite.internal.binary.BinaryUtils;
import org.apache.ignite.internal.processors.cache.persistence.wal.reader.StandaloneGridKernalContext;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
@@ -98,7 +99,7 @@ public class ConcurrentMappingFileReadWriteTest extends GridCommonAbstractTest {
public void testRewriteOpenedFile() throws Exception {
mappingFileStore.writeMapping(PLATFORM_ID, TYPE_ID, String.class.getName());
- File mappingFile = new File(mappingDir, mappingFileStore.getFileName(PLATFORM_ID, TYPE_ID));
+ File mappingFile = new File(mappingDir, BinaryUtils.mappingFileName(PLATFORM_ID, TYPE_ID));
try (FileInputStream in = new FileInputStream(mappingFile)) {
try (BufferedReader reader = new BufferedReader(new InputStreamReader(in, StandardCharsets.UTF_8))) {
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsBinaryMetadataOnClusterRestartTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsBinaryMetadataOnClusterRestartTest.java
index 4645524ec41..f5573749e5e 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsBinaryMetadataOnClusterRestartTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsBinaryMetadataOnClusterRestartTest.java
@@ -39,6 +39,7 @@ import org.apache.ignite.configuration.DataRegionConfiguration;
import org.apache.ignite.configuration.DataStorageConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.configuration.WALMode;
+import org.apache.ignite.internal.binary.BinaryUtils;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.testframework.GridTestUtils;
@@ -332,7 +333,7 @@ public class IgnitePdsBinaryMetadataOnClusterRestartTest extends GridCommonAbstr
cConsId,
String.format(CUSTOM_WORK_DIR_NAME_PATTERN, "B"),
bConsId,
- DYNAMIC_TYPE_NAME.toLowerCase().hashCode() + ".bin");
+ BinaryUtils.binaryMetaFileName(DYNAMIC_TYPE_NAME.toLowerCase().hashCode()));
startGridInASeparateWorkDir("A");
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/cdc/SqlCdcTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/cdc/SqlCdcTest.java
index 83e8ffc2b54..db33d1cd9be 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/cdc/SqlCdcTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/cdc/SqlCdcTest.java
@@ -17,12 +17,17 @@
package org.apache.ignite.internal.cdc;
+import java.util.Arrays;
+import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CountDownLatch;
+import org.apache.ignite.binary.BinaryBasicIdMapper;
import org.apache.ignite.binary.BinaryObject;
+import org.apache.ignite.binary.BinaryType;
import org.apache.ignite.cache.query.SqlFieldsQuery;
import org.apache.ignite.cdc.AbstractCdcTest;
import org.apache.ignite.cdc.CdcEvent;
+import org.apache.ignite.cdc.TypeMapping;
import org.apache.ignite.configuration.DataRegionConfiguration;
import org.apache.ignite.configuration.DataStorageConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
@@ -58,6 +63,15 @@ public class SqlCdcTest extends AbstractCdcTest {
/** */
public static final String MSK = "Moscow";
+ /** */
+ public static final String USER_KEY_TYPE = "TestUserKey";
+
+ /** */
+ public static final String USER_VAL_TYPE = "TestUser";
+
+ /** */
+ public static final String CITY_VAL_TYPE = "TestCity";
+
/** */
@Parameterized.Parameter
public boolean persistenceEnabled;
@@ -103,12 +117,14 @@ public class SqlCdcTest extends AbstractCdcTest {
executeSql(
ign,
- "CREATE TABLE USER(id int, city_id int, name varchar, PRIMARY KEY (id, city_id)) WITH \"CACHE_NAME=user\""
+ "CREATE TABLE USER(id int, city_id int, name varchar, PRIMARY KEY (id, city_id)) " +
+ "WITH \"CACHE_NAME=user,VALUE_TYPE=" + USER_VAL_TYPE + ",KEY_TYPE=" + USER_KEY_TYPE + "\""
);
executeSql(
ign,
- "CREATE TABLE CITY(id int, name varchar, zip_code varchar(6), PRIMARY KEY (id)) WITH \"CACHE_NAME=city\""
+ "CREATE TABLE CITY(id int, name varchar, zip_code varchar(6), PRIMARY KEY (id)) " +
+ "WITH \"CACHE_NAME=city,VALUE_TYPE=TestCity\""
);
for (int i = 0; i < KEYS_CNT; i++) {
@@ -128,7 +144,7 @@ public class SqlCdcTest extends AbstractCdcTest {
}
// Wait while both predicte will become true and state saved on the disk.
- assertTrue(latch.await(getTestTimeout(), MILLISECONDS));;
+ assertTrue(latch.await(getTestTimeout(), MILLISECONDS));
checkMetrics(cdc, KEYS_CNT * 2);
@@ -150,6 +166,18 @@ public class SqlCdcTest extends AbstractCdcTest {
checkMetrics(cdc, KEYS_CNT);
+ executeSql(ign, "ALTER TABLE CITY ADD COLUMN region VARCHAR");
+
+ executeSql(
+ ign,
+ "INSERT INTO CITY VALUES(?, ?, ?, ?)",
+ KEYS_CNT + 1,
+ MSK,
+ Integer.toString(127000 + KEYS_CNT + 1),
+ "Moscow region");
+
+ waitForSize(KEYS_CNT + 1, CITY, UPDATE, cnsmr);
+
rmvFut.cancel();
assertTrue(cnsmr.stopped());
@@ -157,8 +185,24 @@ public class SqlCdcTest extends AbstractCdcTest {
/** */
public static class BinaryCdcConsumer extends TestCdcConsumer<CdcEvent> {
+ /** */
+ private boolean userKeyType;
+
+ /** */
+ private boolean userValType;
+
+ /** */
+ private boolean cityValType;
+
+ /** */
+ private int mappingCnt;
+
/** {@inheritDoc} */
@Override public void checkEvent(CdcEvent evt) {
+ assertTrue(userKeyType);
+ assertTrue(userValType);
+ assertTrue(cityValType);
+
if (evt.value() == null)
return;
@@ -193,6 +237,77 @@ public class SqlCdcTest extends AbstractCdcTest {
@Override public CdcEvent extract(CdcEvent evt) {
return evt;
}
+
+ /** {@inheritDoc} */
+ @Override public void onTypes(Iterator<BinaryType> types) {
+ assertEquals("onMappings must be executed first", 3, mappingCnt);
+
+ while (types.hasNext()) {
+ BinaryType type = types.next();
+
+ assertNotNull(type);
+
+ switch (type.typeName()) {
+ case USER_KEY_TYPE:
+ assertTrue(type.fieldNames().containsAll(Arrays.asList("ID", "CITY_ID")));
+ assertEquals(2, type.fieldNames().size());
+ assertEquals(int.class.getSimpleName(), type.fieldTypeName("ID"));
+ assertEquals(int.class.getSimpleName(), type.fieldTypeName("CITY_ID"));
+
+ userKeyType = true;
+
+ break;
+
+ case USER_VAL_TYPE:
+ assertTrue(type.fieldNames().contains("NAME"));
+ assertEquals(1, type.fieldNames().size());
+ assertEquals(String.class.getSimpleName(), type.fieldTypeName("NAME"));
+
+ userValType = true;
+
+ break;
+
+ case CITY_VAL_TYPE:
+ assertTrue(type.fieldNames().containsAll(Arrays.asList("NAME", "ZIP_CODE")));
+ assertEquals(cityValType ? 3 : 2, type.fieldNames().size());
+ assertEquals(String.class.getSimpleName(), type.fieldTypeName("NAME"));
+ assertEquals(String.class.getSimpleName(), type.fieldTypeName("ZIP_CODE"));
+
+ // Alter table happen.
+ if (cityValType) {
+ assertTrue(type.fieldNames().contains("REGION"));
+ assertEquals(String.class.getSimpleName(), type.fieldTypeName("REGION"));
+ }
+
+ cityValType = true;
+
+ break;
+ default:
+ fail("Unexpected type name " + type.typeName());
+ }
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onMappings(Iterator<TypeMapping> mappings) {
+ assertEquals(0, mappingCnt);
+ assertFalse("onMappings must be executed first", cityValType || userValType || userKeyType);
+
+ BinaryBasicIdMapper mapper = new BinaryBasicIdMapper();
+
+ while (mappings.hasNext()) {
+ mappingCnt++;
+
+ TypeMapping m = mappings.next();
+
+ assertNotNull(m);
+
+ String typeName = m.typeName();
+
+ assertFalse(typeName.isEmpty());
+ assertEquals(mapper.typeId(typeName), m.typeId());
+ }
+ }
}
/** */
diff --git a/modules/spring/src/test/java/org/apache/ignite/cdc/CdcConfigurationTest.java b/modules/spring/src/test/java/org/apache/ignite/cdc/CdcConfigurationTest.java
index 53e10d90bf4..84ac26bcf88 100644
--- a/modules/spring/src/test/java/org/apache/ignite/cdc/CdcConfigurationTest.java
+++ b/modules/spring/src/test/java/org/apache/ignite/cdc/CdcConfigurationTest.java
@@ -22,6 +22,7 @@ import java.util.concurrent.CountDownLatch;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.binary.BinaryType;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInternalFuture;
@@ -133,6 +134,16 @@ public class CdcConfigurationTest extends GridCommonAbstractTest {
return false;
}
+ /** {@inheritDoc} */
+ @Override public void onTypes(Iterator<BinaryType> types) {
+ // No-Op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onMappings(Iterator<TypeMapping> mappings) {
+ // No-Op.
+ }
+
/** {@inheritDoc} */
@Override public void stop() {
// No-Op.