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/19 06:29:39 UTC
[ignite-extensions] branch master updated: IGNITE-14449 CDC Metadata replication added.
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-extensions.git
The following commit(s) were added to refs/heads/master by this push:
new 57e14df IGNITE-14449 CDC Metadata replication added.
57e14df is described below
commit 57e14df79b44de6fe9c69984042179acf9e2d252
Author: Nikolay Izhikov <ni...@apache.org>
AuthorDate: Tue Apr 19 09:29:21 2022 +0300
IGNITE-14449 CDC Metadata replication added.
---
.../org/apache/ignite/cdc/CdcEventsApplier.java | 133 ++++++++++++++++-
.../ignite/cdc/IgniteToIgniteCdcStreamer.java | 47 ++++++
.../ignite/cdc/kafka/IgniteToKafkaCdcStreamer.java | 165 ++++++++++++++++-----
.../ignite/cdc/kafka/KafkaToIgniteCdcStreamer.java | 46 ++++--
.../cdc/kafka/KafkaToIgniteCdcStreamerApplier.java | 25 +++-
.../KafkaToIgniteCdcStreamerConfiguration.java | 70 ++++++++-
.../cdc/kafka/KafkaToIgniteMetadataUpdater.java | 152 +++++++++++++++++++
.../apache/ignite/cdc/AbstractReplicationTest.java | 9 --
.../cdc/kafka/CdcKafkaReplicationAppsTest.java | 13 +-
.../ignite/cdc/kafka/CdcKafkaReplicationTest.java | 47 +++++-
.../resources/loader/kafka-to-ignite-correct.xml | 1 +
.../test/resources/replication/ignite-to-kafka.xml | 1 +
.../test/resources/replication/kafka-to-ignite.xml | 1 +
13 files changed, 623 insertions(+), 87 deletions(-)
diff --git a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsApplier.java b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsApplier.java
index 1751ed6..c76c717 100644
--- a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsApplier.java
+++ b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsApplier.java
@@ -21,18 +21,27 @@ import java.util.HashMap;
import java.util.Map;
import java.util.function.BooleanSupplier;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.binary.BinaryObject;
+import org.apache.ignite.binary.BinaryObjectException;
import org.apache.ignite.cache.CacheEntryVersion;
+import org.apache.ignite.cdc.kafka.KafkaToIgniteCdcStreamerConfiguration;
+import org.apache.ignite.cdc.kafka.KafkaToIgniteMetadataUpdater;
import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.binary.BinaryMetadata;
+import org.apache.ignite.internal.binary.BinaryTypeImpl;
import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.CacheObjectImpl;
import org.apache.ignite.internal.processors.cache.IgniteInternalCache;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.KeyCacheObjectImpl;
+import org.apache.ignite.internal.processors.cache.binary.CacheObjectBinaryProcessorImpl;
import org.apache.ignite.internal.processors.cache.dr.GridCacheDrInfo;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.util.lang.IgniteInClosureX;
import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.CU;
/**
@@ -57,6 +66,34 @@ public abstract class CdcEventsApplier {
/** */
private final BooleanSupplier hasRemoves = () -> !F.isEmpty(rmvBatch);
+ /**
+ * Update closure.
+ * @see #applyWithRetry(IgniteInClosureX, IgniteInternalCache)
+ */
+ private final IgniteInClosureX<IgniteInternalCache<BinaryObject, BinaryObject>> updClo =
+ new IgniteInClosureX<IgniteInternalCache<BinaryObject, BinaryObject>>() {
+ /** {@inheritDoc} */
+ @Override public void applyx(
+ IgniteInternalCache<BinaryObject, BinaryObject> cache
+ ) throws IgniteCheckedException {
+ cache.putAllConflict(updBatch);
+ }
+ };
+
+ /**
+ * Remove closure.
+ * @see #applyWithRetry(IgniteInClosureX, IgniteInternalCache)
+ */
+ private final IgniteInClosureX<IgniteInternalCache<BinaryObject, BinaryObject>> rmvClo =
+ new IgniteInClosureX<IgniteInternalCache<BinaryObject, BinaryObject>>() {
+ /** {@inheritDoc} */
+ @Override public void applyx(
+ IgniteInternalCache<BinaryObject, BinaryObject> cache
+ ) throws IgniteCheckedException {
+ cache.removeAllConflict(rmvBatch);
+ }
+ };
+
/**
* @param maxBatchSize Maximum batch size.
*/
@@ -84,7 +121,11 @@ public abstract class CdcEventsApplier {
// IgniteEx#cachex(String) will return null if cache not initialized with regular Ignite#cache(String) call.
ignite().cache(cacheName);
- return ignite().cachex(cacheName).keepBinary();
+ IgniteInternalCache<Object, Object> cache0 = ignite().cachex(cacheName);
+
+ assert cache0 != null;
+
+ return cache0.keepBinary();
}
}
@@ -140,20 +181,19 @@ public abstract class CdcEventsApplier {
* @param applyUpd Apply update batch flag supplier.
* @param applyRmv Apply remove batch flag supplier.
* @return Number of applied events.
- * @throws IgniteCheckedException In case of error.
*/
private int applyIf(
IgniteInternalCache<BinaryObject, BinaryObject> cache,
BooleanSupplier applyUpd,
BooleanSupplier applyRmv
- ) throws IgniteCheckedException {
+ ) {
int evtsApplied = 0;
if (applyUpd.getAsBoolean()) {
if (log().isDebugEnabled())
log().debug("Applying put batch [cache=" + cache.name() + ']');
- cache.putAllConflict(updBatch);
+ applyWithRetry(updClo, cache);
evtsApplied += updBatch.size();
@@ -164,7 +204,7 @@ public abstract class CdcEventsApplier {
if (log().isDebugEnabled())
log().debug("Applying remove batch [cache=" + cache.name() + ']');
- cache.removeAllConflict(rmvBatch);
+ applyWithRetry(rmvClo, cache);
evtsApplied += rmvBatch.size();
@@ -174,11 +214,94 @@ public abstract class CdcEventsApplier {
return evtsApplied;
}
+ /**
+ * Executes closure with retry logic.
+ * Metadata update thread polls metadata asynchronously with {@link KafkaToIgniteCdcStreamerConfiguration#getMetaUpdateInterval()} interval.
+ * This means metadata updates can be seen later than data updates.
+ * In this case {@link BinaryObjectException} can point to absence of metadata.
+ * To overcome lack of metadata invoke {@link #updateMetadata()} and retry closure.
+ *
+ * @param clo Closure to apply.
+ * @param cache Cache for closure.
+ * @see KafkaToIgniteMetadataUpdater
+ * @see KafkaToIgniteCdcStreamerConfiguration#getMetaUpdateInterval()
+ */
+ private void applyWithRetry(
+ IgniteInClosureX<IgniteInternalCache<BinaryObject, BinaryObject>> clo,
+ IgniteInternalCache<BinaryObject, BinaryObject> cache
+ ) {
+ try {
+ clo.apply(cache);
+ }
+ catch (Exception e) {
+ // Retry only if cause is BinaryObjectException.
+ if (!X.hasCause(e, BinaryObjectException.class))
+ throw e;
+
+ // Retry after metadata update.
+ updateMetadata();
+
+ clo.apply(cache);
+ }
+ }
+
/** @return {@code True} if update batch should be applied. */
private boolean isApplyBatch(Map<KeyCacheObject, ?> map, KeyCacheObject key) {
return map.size() >= maxBatchSize || map.containsKey(key);
}
+ /**
+ * Register {@code meta} inside {@code ign} instance.
+ *
+ * @param ign Ignite instance.
+ * @param log Logger.
+ * @param meta Binary metadata to register.
+ */
+ public static void registerBinaryMeta(IgniteEx ign, IgniteLogger log, BinaryMetadata meta) {
+ ign.context().cacheObjects().addMeta(
+ meta.typeId(),
+ new BinaryTypeImpl(
+ ((CacheObjectBinaryProcessorImpl)ign.context().cacheObjects()).binaryContext(),
+ meta
+ ),
+ false
+ );
+
+ if (log.isInfoEnabled())
+ log.info("BinaryMeta[meta=" + meta + ']');
+ }
+
+ /**
+ * Register {@code mapping} inside {@code ign} instance.
+ *
+ * @param ign Ignite instance.
+ * @param log Logger.
+ * @param mapping Type mapping to register.
+ */
+ public static void registerMapping(IgniteEx ign, IgniteLogger log, TypeMapping mapping) {
+ assert mapping.platformType().ordinal() <= Byte.MAX_VALUE;
+
+ try {
+ ign.context().marshallerContext().registerClassName(
+ (byte)mapping.platformType().ordinal(),
+ mapping.typeId(),
+ mapping.typeName(),
+ false
+ );
+
+ if (log.isInfoEnabled())
+ log.info("Mapping[mapping=" + mapping + ']');
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteException(e);
+ }
+ }
+
+ /** Update metadata if possible. */
+ protected void updateMetadata() {
+ // No-op.
+ }
+
/** @return Ignite instance. */
protected abstract IgniteEx ignite();
diff --git a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/IgniteToIgniteCdcStreamer.java b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/IgniteToIgniteCdcStreamer.java
index 81cd892..780c8c8 100644
--- a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/IgniteToIgniteCdcStreamer.java
+++ b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/IgniteToIgniteCdcStreamer.java
@@ -24,10 +24,13 @@ import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.Ignition;
+import org.apache.ignite.binary.BinaryType;
import org.apache.ignite.cdc.conflictresolve.CacheVersionConflictResolverImpl;
import org.apache.ignite.cdc.kafka.KafkaToIgniteCdcStreamer;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.binary.BinaryMetadata;
+import org.apache.ignite.internal.binary.BinaryTypeImpl;
import org.apache.ignite.internal.cdc.CdcMain;
import org.apache.ignite.internal.processors.metric.MetricRegistry;
import org.apache.ignite.internal.processors.metric.impl.AtomicLongMetric;
@@ -59,9 +62,21 @@ public class IgniteToIgniteCdcStreamer extends CdcEventsApplier implements CdcCo
/** */
public static final String EVTS_CNT = "EventsCount";
+ /** */
+ public static final String TYPES_CNT = "TypesCount";
+
+ /** */
+ public static final String MAPPINGS_CNT = "MappingsCount";
+
/** */
public static final String EVTS_CNT_DESC = "Count of messages applied to destination cluster";
+ /** */
+ public static final String TYPES_CNT_DESC = "Count of received binary types events";
+
+ /** */
+ public static final String MAPPINGS_CNT_DESC = "Count of received mappings events";
+
/** */
public static final String LAST_EVT_TIME = "LastEventTime";
@@ -89,6 +104,12 @@ public class IgniteToIgniteCdcStreamer extends CdcEventsApplier implements CdcCo
/** Count of events applied to destination cluster. */
protected AtomicLongMetric evtsCnt;
+ /** Count of binary types applied to destination cluster. */
+ protected AtomicLongMetric typesCnt;
+
+ /** Count of mappings applied to destination cluster. */
+ protected AtomicLongMetric mappingsCnt;
+
/** Logger. */
@LoggerResource
private IgniteLogger log;
@@ -114,6 +135,8 @@ public class IgniteToIgniteCdcStreamer extends CdcEventsApplier implements CdcCo
dest = (IgniteEx)Ignition.start(destIgniteCfg);
this.evtsCnt = mreg.longMetric(EVTS_CNT, EVTS_CNT_DESC);
+ this.typesCnt = mreg.longMetric(TYPES_CNT, TYPES_CNT_DESC);
+ this.mappingsCnt = mreg.longMetric(MAPPINGS_CNT, MAPPINGS_CNT_DESC);
this.lastEvtTs = mreg.longMetric(LAST_EVT_TIME, LAST_EVT_TIME_DESC);
}
@@ -143,6 +166,30 @@ public class IgniteToIgniteCdcStreamer extends CdcEventsApplier implements CdcCo
}
}
+ /** {@inheritDoc} */
+ @Override public void onTypes(Iterator<BinaryType> types) {
+ types.forEachRemaining(t -> {
+ BinaryMetadata meta = ((BinaryTypeImpl)t).metadata();
+
+ registerBinaryMeta(dest, log, meta);
+
+ typesCnt.increment();
+ });
+
+ lastEvtTs.value(System.currentTimeMillis());
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onMappings(Iterator<TypeMapping> mappings) {
+ mappings.forEachRemaining(m -> {
+ registerMapping(dest, log, m);
+
+ mappingsCnt.increment();
+ });
+
+ lastEvtTs.value(System.currentTimeMillis());
+ }
+
/** {@inheritDoc} */
@Override public void stop() {
dest.close();
diff --git a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/IgniteToKafkaCdcStreamer.java b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/IgniteToKafkaCdcStreamer.java
index be80532..9c1b9c5 100644
--- a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/IgniteToKafkaCdcStreamer.java
+++ b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/IgniteToKafkaCdcStreamer.java
@@ -27,15 +27,20 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.function.Function;
import java.util.stream.Collectors;
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.cdc.conflictresolve.CacheVersionConflictResolverImpl;
+import org.apache.ignite.internal.binary.BinaryTypeImpl;
import org.apache.ignite.internal.cdc.CdcMain;
import org.apache.ignite.internal.processors.metric.MetricRegistry;
import org.apache.ignite.internal.processors.metric.impl.AtomicLongMetric;
import org.apache.ignite.internal.util.IgniteUtils;
+import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.A;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.lang.IgniteExperimental;
@@ -50,6 +55,10 @@ import static org.apache.ignite.cdc.IgniteToIgniteCdcStreamer.EVTS_CNT;
import static org.apache.ignite.cdc.IgniteToIgniteCdcStreamer.EVTS_CNT_DESC;
import static org.apache.ignite.cdc.IgniteToIgniteCdcStreamer.LAST_EVT_TIME;
import static org.apache.ignite.cdc.IgniteToIgniteCdcStreamer.LAST_EVT_TIME_DESC;
+import static org.apache.ignite.cdc.IgniteToIgniteCdcStreamer.MAPPINGS_CNT;
+import static org.apache.ignite.cdc.IgniteToIgniteCdcStreamer.MAPPINGS_CNT_DESC;
+import static org.apache.ignite.cdc.IgniteToIgniteCdcStreamer.TYPES_CNT;
+import static org.apache.ignite.cdc.IgniteToIgniteCdcStreamer.TYPES_CNT_DESC;
import static org.apache.ignite.cdc.kafka.KafkaToIgniteCdcStreamerConfiguration.DFLT_KAFKA_REQ_TIMEOUT;
import static org.apache.ignite.cdc.kafka.KafkaToIgniteCdcStreamerConfiguration.DFLT_MAX_BATCH_SIZE;
import static org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG;
@@ -86,14 +95,17 @@ public class IgniteToKafkaCdcStreamer implements CdcConsumer {
@LoggerResource
private IgniteLogger log;
- /** Kafka producer to stream events. */
+ /** Kafka producer. */
private KafkaProducer<Integer, byte[]> producer;
/** Handle only primary entry flag. */
private boolean onlyPrimary = DFLT_IS_ONLY_PRIMARY;
- /** Topic name. */
- private String topic;
+ /** Topic to send data. */
+ private String evtTopic;
+
+ /** Topic to send metadata. */
+ private String metadataTopic;
/** Kafka topic partitions count. */
private int kafkaParts;
@@ -108,7 +120,7 @@ public class IgniteToKafkaCdcStreamer implements CdcConsumer {
private Collection<String> caches;
/** Max batch size. */
- private int maxBatchSize = DFLT_MAX_BATCH_SIZE;
+ private int maxBatchSz = DFLT_MAX_BATCH_SIZE;
/** The maximum time to complete Kafka related requests, in milliseconds. */
private long kafkaReqTimeout = DFLT_KAFKA_REQ_TIMEOUT;
@@ -119,16 +131,22 @@ public class IgniteToKafkaCdcStreamer implements CdcConsumer {
/** Count of bytes sent to the Kafka. */
private AtomicLongMetric bytesSnt;
- /** Count of sent messages. */
- private AtomicLongMetric msgsSnt;
+ /** Count of sent events. */
+ private AtomicLongMetric evtsCnt;
- /** {@inheritDoc} */
- @Override public boolean onEvents(Iterator<CdcEvent> evts) {
- List<Future<RecordMetadata>> futs = new ArrayList<>();
+ /** Count of sent binary types. */
+ protected AtomicLongMetric typesCnt;
+
+ /** Count of sent mappings. */
+ protected AtomicLongMetric mappingsCnt;
+
+ /** */
+ private List<Future<RecordMetadata>> futs;
- while (evts.hasNext() && futs.size() < maxBatchSize) {
- CdcEvent evt = evts.next();
+ /** {@inheritDoc} */
+ @Override public boolean onEvents(Iterator<CdcEvent> evts) {
+ Iterator<CdcEvent> filtered = F.iterator(evts, e -> e, true, evt -> {
if (log.isDebugEnabled())
log.debug("Event received [evt=" + evt + ']');
@@ -136,7 +154,7 @@ public class IgniteToKafkaCdcStreamer implements CdcConsumer {
if (log.isDebugEnabled())
log.debug("Event skipped because of primary flag [evt=" + evt + ']');
- continue;
+ return false;
}
if (evt.version().otherClusterVersion() != null) {
@@ -145,29 +163,74 @@ public class IgniteToKafkaCdcStreamer implements CdcConsumer {
", otherClusterVersion=" + evt.version().otherClusterVersion() + ']');
}
- continue;
+ return false;
}
if (!cachesIds.contains(evt.cacheId())) {
if (log.isDebugEnabled())
log.debug("Event skipped because of cacheId [evt=" + evt + ']');
- continue;
+ return false;
}
- byte[] bytes = IgniteUtils.toBytes(evt);
+ return true;
+ });
+
+ while (filtered.hasNext()) {
+ sendLimited(
+ filtered,
+ evt -> new ProducerRecord<>(
+ evtTopic,
+ evt.partition() % kafkaParts,
+ evt.cacheId(),
+ IgniteUtils.toBytes(evt)
+ ),
+ evtsCnt
+ );
+ }
+
+ return true;
+ }
- bytesSnt.add(bytes.length);
+ /** {@inheritDoc} */
+ @Override public void onTypes(Iterator<BinaryType> types) {
+ while (types.hasNext()) {
+ sendLimited(
+ types,
+ t -> new ProducerRecord<>(metadataTopic, IgniteUtils.toBytes(((BinaryTypeImpl)t).metadata())),
+ typesCnt
+ );
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onMappings(Iterator<TypeMapping> mappings) {
+ while (mappings.hasNext()) {
+ sendLimited(
+ mappings,
+ m -> new ProducerRecord<>(metadataTopic, IgniteUtils.toBytes(m)),
+ mappingsCnt
+ );
+ }
+ }
- futs.add(producer.send(new ProducerRecord<>(
- topic,
- evt.partition() % kafkaParts,
- evt.cacheId(),
- bytes
- )));
+ /** Send limited amount of data to Kafka. */
+ private <T> void sendLimited(
+ Iterator<T> data,
+ Function<T, ProducerRecord<Integer, byte[]>> toRec,
+ AtomicLongMetric cntr
+ ) {
+ while (data.hasNext() && futs.size() < maxBatchSz) {
+ T item = data.next();
if (log.isDebugEnabled())
- log.debug("Event sent asynchronously [evt=" + evt + ']');
+ log.debug("Sent asynchronously [item=" + item + ']');
+
+ ProducerRecord<Integer, byte[]> rec = toRec.apply(item);
+
+ bytesSnt.add(rec.value().length);
+
+ futs.add(producer.send(rec));
}
if (!futs.isEmpty()) {
@@ -175,25 +238,25 @@ public class IgniteToKafkaCdcStreamer implements CdcConsumer {
for (Future<RecordMetadata> fut : futs)
fut.get(kafkaReqTimeout, TimeUnit.MILLISECONDS);
- msgsSnt.add(futs.size());
-
+ cntr.add(futs.size());
lastMsgTs.value(System.currentTimeMillis());
+
+ futs.clear();
}
catch (InterruptedException | ExecutionException | TimeoutException e) {
throw new RuntimeException(e);
}
if (log.isInfoEnabled())
- log.info("Events processed [sentMessagesCount=" + msgsSnt.value() + ']');
+ log.info("Items processed [count=" + cntr.value() + ']');
}
-
- return true;
}
/** {@inheritDoc} */
@Override public void start(MetricRegistry mreg) {
A.notNull(kafkaProps, "Kafka properties");
- A.notNull(topic, "Kafka topic");
+ A.notNull(evtTopic, "Kafka topic");
+ A.notNull(metadataTopic, "Kafka metadata topic");
A.notEmpty(caches, "caches");
A.ensure(kafkaParts > 0, "The number of Kafka partitions must be explicitly set to a value greater than zero.");
A.ensure(kafkaReqTimeout >= 0, "The Kafka request timeout cannot be negative.");
@@ -208,16 +271,26 @@ public class IgniteToKafkaCdcStreamer implements CdcConsumer {
try {
producer = new KafkaProducer<>(kafkaProps);
- if (log.isInfoEnabled())
- log.info("CDC Ignite To Kafka started [topic=" + topic + ", onlyPrimary=" + onlyPrimary + ", cacheIds=" + cachesIds + ']');
+ if (log.isInfoEnabled()) {
+ log.info("CDC Ignite To Kafka started [" +
+ "topic=" + evtTopic +
+ ", metadataTopic = " + metadataTopic +
+ ", onlyPrimary=" + onlyPrimary +
+ ", cacheIds=" + cachesIds + ']'
+ );
+ }
}
catch (Exception e) {
throw new RuntimeException(e);
}
- this.msgsSnt = mreg.longMetric(EVTS_CNT, EVTS_CNT_DESC);
+ this.evtsCnt = mreg.longMetric(EVTS_CNT, EVTS_CNT_DESC);
this.lastMsgTs = mreg.longMetric(LAST_EVT_TIME, LAST_EVT_TIME_DESC);
this.bytesSnt = mreg.longMetric(BYTES_SENT, BYTES_SENT_DESCRIPTION);
+ this.typesCnt = mreg.longMetric(TYPES_CNT, TYPES_CNT_DESC);
+ this.mappingsCnt = mreg.longMetric(MAPPINGS_CNT, MAPPINGS_CNT_DESC);
+
+ futs = new ArrayList<>(maxBatchSz);
}
/** {@inheritDoc} */
@@ -240,19 +313,31 @@ public class IgniteToKafkaCdcStreamer implements CdcConsumer {
/**
* Sets topic that is used to send data to Kafka.
*
- * @param topic Kafka topic.
+ * @param evtTopic Kafka topic.
+ * @return {@code this} for chaining.
+ */
+ public IgniteToKafkaCdcStreamer setTopic(String evtTopic) {
+ this.evtTopic = evtTopic;
+
+ return this;
+ }
+
+ /**
+ * Sets topic that is used to send metadata to Kafka.
+ *
+ * @param metadataTopic Metadata topic.
* @return {@code this} for chaining.
*/
- public IgniteToKafkaCdcStreamer setTopic(String topic) {
- this.topic = topic;
+ public IgniteToKafkaCdcStreamer setMetadataTopic(String metadataTopic) {
+ this.metadataTopic = metadataTopic;
return this;
}
/**
- * Sets number of Kafka partitions.
+ * Sets number of Kafka partitions for data topic.
*
- * @param kafkaParts Number of Kafka partitions.
+ * @param kafkaParts Number of Kafka partitions for data topic.
* @return {@code this} for chaining.
*/
public IgniteToKafkaCdcStreamer setKafkaPartitions(int kafkaParts) {
@@ -276,11 +361,11 @@ public class IgniteToKafkaCdcStreamer implements CdcConsumer {
/**
* Sets maximum batch size.
*
- * @param maxBatchSize Maximum batch size.
+ * @param maxBatchSz Maximum batch size.
* @return {@code this} for chaining.
*/
- public IgniteToKafkaCdcStreamer setMaxBatchSize(int maxBatchSize) {
- this.maxBatchSize = maxBatchSize;
+ public IgniteToKafkaCdcStreamer setMaxBatchSize(int maxBatchSz) {
+ this.maxBatchSz = maxBatchSz;
return this;
}
diff --git a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteCdcStreamer.java b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteCdcStreamer.java
index a0d3880..3db02ba 100644
--- a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteCdcStreamer.java
+++ b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteCdcStreamer.java
@@ -97,10 +97,10 @@ public class KafkaToIgniteCdcStreamer implements Runnable {
private final KafkaToIgniteCdcStreamerConfiguration streamerCfg;
/** Runners to run {@link KafkaToIgniteCdcStreamerApplier} instances. */
- private final Thread[] runners;
+ private final List<Thread> runners;
/** Appliers. */
- private final List<KafkaToIgniteCdcStreamerApplier> appliers;
+ private final List<AutoCloseable> appliers;
/**
* @param igniteCfg Ignite configuration.
@@ -113,6 +113,7 @@ public class KafkaToIgniteCdcStreamer implements Runnable {
KafkaToIgniteCdcStreamerConfiguration streamerCfg
) {
A.notNull(streamerCfg.getTopic(), "Kafka topic");
+ A.notNull(streamerCfg.getMetadataTopic(), "Kafka metadata topic");
A.ensure(
streamerCfg.getKafkaPartsFrom() >= 0,
"The Kafka partitions lower bound must be explicitly set to a value greater than or equals to zero.");
@@ -132,8 +133,9 @@ public class KafkaToIgniteCdcStreamer implements Runnable {
this.kafkaProps = kafkaProps;
this.streamerCfg = streamerCfg;
- appliers = new ArrayList<>(streamerCfg.getThreadCount());
- runners = new Thread[streamerCfg.getThreadCount()];
+ // Extra thread for metadata updater.
+ appliers = new ArrayList<>(streamerCfg.getThreadCount() + 1);
+ runners = new ArrayList<>(streamerCfg.getThreadCount() + 1);
if (!kafkaProps.containsKey(ConsumerConfig.GROUP_ID_CONFIG))
throw new IllegalArgumentException("Kafka properties don't contains " + ConsumerConfig.GROUP_ID_CONFIG);
@@ -173,6 +175,16 @@ public class KafkaToIgniteCdcStreamer implements Runnable {
.map(CU::cacheId).collect(Collectors.toSet());
}
+ KafkaToIgniteMetadataUpdater metaUpdr = new KafkaToIgniteMetadataUpdater(
+ ign,
+ log,
+ kafkaProps,
+ streamerCfg,
+ stopped
+ );
+
+ addAndStart("meta-update-thread", metaUpdr);
+
int kafkaPartsFrom = streamerCfg.getKafkaPartsFrom();
int kafkaParts = streamerCfg.getKafkaPartsTo() - kafkaPartsFrom;
int threadCnt = streamerCfg.getThreadCount();
@@ -195,20 +207,17 @@ public class KafkaToIgniteCdcStreamer implements Runnable {
kafkaPartsFrom + to,
caches,
streamerCfg.getMaxBatchSize(),
- stopped,
- streamerCfg.getKafkaRequestTimeout()
+ streamerCfg.getKafkaRequestTimeout(),
+ metaUpdr,
+ stopped
);
- appliers.add(applier);
-
- runners[i] = new Thread(applier, "applier-thread-" + i);
-
- runners[i].start();
+ addAndStart("applier-thread-" + i, applier);
}
try {
- for (int i = 0; i < threadCnt; i ++)
- runners[i].join();
+ for (int i = 0; i < threadCnt + 1; i ++)
+ runners.get(i).join();
}
catch (InterruptedException e) {
stopped.set(true);
@@ -220,6 +229,17 @@ public class KafkaToIgniteCdcStreamer implements Runnable {
}
}
+ /** Adds applier to {@link #appliers} and starts thread with it. */
+ private <T extends AutoCloseable & Runnable> void addAndStart(String threadName, T applier) {
+ appliers.add(applier);
+
+ Thread thread = new Thread(applier, threadName);
+
+ thread.start();
+
+ runners.add(thread);
+ }
+
/** */
private void ackAsciiLogo(IgniteLogger log) {
String ver = "ver. " + ACK_VER_STR;
diff --git a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteCdcStreamerApplier.java b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteCdcStreamerApplier.java
index 985deb7..5de8cc3 100644
--- a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteCdcStreamerApplier.java
+++ b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteCdcStreamerApplier.java
@@ -105,15 +105,18 @@ class KafkaToIgniteCdcStreamerApplier extends CdcEventsApplier implements Runnab
/** Caches ids to read. */
private final Set<Integer> caches;
+ /** The maximum time to complete Kafka related requests, in milliseconds. */
+ private final long kafkaReqTimeout;
+
+ /** Metadata updater. */
+ private final KafkaToIgniteMetadataUpdater metaUpdr;
+
/** Consumers. */
private final List<KafkaConsumer<Integer, byte[]>> cnsmrs = new ArrayList<>();
/** */
private final AtomicLong rcvdEvts = new AtomicLong();
- /** The maximum time to complete Kafka related requests, in milliseconds. */
- private final long kafkaReqTimeout;
-
/**
* @param ign Ignite instance.
* @param log Logger.
@@ -123,8 +126,9 @@ class KafkaToIgniteCdcStreamerApplier extends CdcEventsApplier implements Runnab
* @param kafkaPartTo Read to partition.
* @param caches Cache ids.
* @param maxBatchSize Maximum batch size.
- * @param stopped Stopped flag.
* @param kafkaReqTimeout The maximum time to complete Kafka related requests, in milliseconds.
+ * @param metaUpdr Metadata updater.
+ * @param stopped Stopped flag.
*/
public KafkaToIgniteCdcStreamerApplier(
IgniteEx ign,
@@ -135,8 +139,9 @@ class KafkaToIgniteCdcStreamerApplier extends CdcEventsApplier implements Runnab
int kafkaPartTo,
Set<Integer> caches,
int maxBatchSize,
- AtomicBoolean stopped,
- long kafkaReqTimeout
+ long kafkaReqTimeout,
+ KafkaToIgniteMetadataUpdater metaUpdr,
+ AtomicBoolean stopped
) {
super(maxBatchSize);
@@ -146,9 +151,10 @@ class KafkaToIgniteCdcStreamerApplier extends CdcEventsApplier implements Runnab
this.kafkaPartFrom = kafkaPartFrom;
this.kafkaPartTo = kafkaPartTo;
this.caches = caches;
+ this.kafkaReqTimeout = kafkaReqTimeout;
+ this.metaUpdr = metaUpdr;
this.stopped = stopped;
this.log = log.getLogger(KafkaToIgniteCdcStreamerApplier.class);
- this.kafkaReqTimeout = kafkaReqTimeout;
}
/** {@inheritDoc} */
@@ -230,6 +236,11 @@ class KafkaToIgniteCdcStreamerApplier extends CdcEventsApplier implements Runnab
}
}
+ /** {@inheritDoc} */
+ @Override protected void updateMetadata() {
+ metaUpdr.updateMetadata();
+ }
+
/** {@inheritDoc} */
@Override public void close() {
log.warning("Close applier!");
diff --git a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteCdcStreamerConfiguration.java b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteCdcStreamerConfiguration.java
index 2a14609..3de9df3 100644
--- a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteCdcStreamerConfiguration.java
+++ b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteCdcStreamerConfiguration.java
@@ -33,6 +33,9 @@ public class KafkaToIgniteCdcStreamerConfiguration {
/** Default maximum time to complete Kafka related requests, in milliseconds. */
public static final long DFLT_KAFKA_REQ_TIMEOUT = 3_000L;
+ /** Default maximum time to complete Kafka related requests, in milliseconds. */
+ public static final long DFLT_META_UPD_INTERVAL = 3_000L;
+
/** Default {@link #threadCnt} value. */
public static final int DFLT_THREAD_CNT = 16;
@@ -42,8 +45,11 @@ public class KafkaToIgniteCdcStreamerConfiguration {
/** {@link KafkaToIgniteCdcStreamerApplier} thread count. */
private int threadCnt = DFLT_THREAD_CNT;
- /** Topic name. */
- private String topic;
+ /** Events topic name. */
+ private String evtTopic;
+
+ /** Metadata topic name. */
+ private String metadataTopic;
/** Kafka partitions lower bound (inclusive). */
private int kafkaPartsFrom = -1;
@@ -54,6 +60,12 @@ public class KafkaToIgniteCdcStreamerConfiguration {
/** The maximum time to complete Kafka related requests, in milliseconds. */
private long kafkaReqTimeout = DFLT_KAFKA_REQ_TIMEOUT;
+ /** Amount of time between two polling of {@link #metadataTopic}, in milliseconds. */
+ private long metaUpdInterval = DFLT_META_UPD_INTERVAL;
+
+ /** Metadata consumer group. */
+ private String metadataCnsmrGrp;
+
/**
* Maximum batch size to apply to Ignite.
*
@@ -79,12 +91,12 @@ public class KafkaToIgniteCdcStreamerConfiguration {
/** */
public String getTopic() {
- return topic;
+ return evtTopic;
}
/** */
- public void setTopic(String topic) {
- this.topic = topic;
+ public void setTopic(String evtTopic) {
+ this.evtTopic = evtTopic;
}
/** */
@@ -140,4 +152,52 @@ public class KafkaToIgniteCdcStreamerConfiguration {
public void setKafkaRequestTimeout(long kafkaReqTimeout) {
this.kafkaReqTimeout = kafkaReqTimeout;
}
+
+ /**
+ * @return Metadata topic name.
+ */
+ public String getMetadataTopic() {
+ return metadataTopic;
+ }
+
+ /**
+ * Sets metadata topic name.
+ *
+ * @param metadataTopic Metadata topic name.
+ */
+ public void setMetadataTopic(String metadataTopic) {
+ this.metadataTopic = metadataTopic;
+ }
+
+ /**
+ * @return Amount of time between two polling of {@link #metadataTopic}.
+ */
+ public long getMetaUpdateInterval() {
+ return metaUpdInterval;
+ }
+
+ /**
+ * Sets amount of time between two polling of {@link #metadataTopic}.
+ *
+ * @param metaUpdateInterval Amount of time between two polling of {@link #metadataTopic}.
+ */
+ public void setMetaUpdateInterval(long metaUpdateInterval) {
+ this.metaUpdInterval = metaUpdateInterval;
+ }
+
+ /**
+ * @return Consumer group to read metadata topic.
+ */
+ public String getMetadataConsumerGroup() {
+ return metadataCnsmrGrp;
+ }
+
+ /**
+ * Sets consumer group to read metadata topic.
+ *
+ * @param metaCnsmrGrp Consumer group to read metadata topic.
+ */
+ public void setMetadataConsumerGroup(String metaCnsmrGrp) {
+ this.metadataCnsmrGrp = metaCnsmrGrp;
+ }
}
diff --git a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteMetadataUpdater.java b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteMetadataUpdater.java
new file mode 100644
index 0000000..ae131db
--- /dev/null
+++ b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteMetadataUpdater.java
@@ -0,0 +1,152 @@
+/*
+ * 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.kafka;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.cdc.CdcEventsApplier;
+import org.apache.ignite.cdc.TypeMapping;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.binary.BinaryMetadata;
+import org.apache.ignite.internal.util.IgniteUtils;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.VoidDeserializer;
+
+import static org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_ID_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG;
+
+public class KafkaToIgniteMetadataUpdater implements AutoCloseable, Runnable {
+ /** Ignite instance. */
+ private final IgniteEx ign;
+
+ /** Log. */
+ private final IgniteLogger log;
+
+ /** Closed flag. Shared between all appliers. */
+ private final AtomicBoolean stopped;
+
+ /** The maximum time to complete Kafka related requests, in milliseconds. */
+ private final long kafkaReqTimeout;
+
+ /** The maximum time to complete Kafka related requests, in milliseconds. */
+ private final long metaUpdInterval;
+
+ /** */
+ private final KafkaConsumer<Void, byte[]> cnsmr;
+
+ /** */
+ private final AtomicLong rcvdEvts = new AtomicLong();
+
+ /**
+ * @param ign Ignite instance.
+ * @param log Logger.
+ * @param initProps Kafka properties.
+ * @param streamerCfg Streamer configuration.
+ * @param stopped Stopped flag.
+ */
+ public KafkaToIgniteMetadataUpdater(
+ IgniteEx ign,
+ IgniteLogger log,
+ Properties initProps,
+ KafkaToIgniteCdcStreamerConfiguration streamerCfg,
+ AtomicBoolean stopped
+ ) {
+ this.ign = ign;
+ this.kafkaReqTimeout = streamerCfg.getKafkaRequestTimeout();
+ this.metaUpdInterval = streamerCfg.getMetaUpdateInterval();
+ this.stopped = stopped;
+ this.log = log.getLogger(KafkaToIgniteMetadataUpdater.class);
+
+ Properties kafkaProps = new Properties();
+
+ kafkaProps.putAll(initProps);
+ kafkaProps.put(KEY_DESERIALIZER_CLASS_CONFIG, VoidDeserializer.class.getName());
+ kafkaProps.put(VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
+ kafkaProps.put(GROUP_ID_CONFIG, streamerCfg.getMetadataConsumerGroup() != null
+ ? streamerCfg.getMetadataConsumerGroup()
+ : ("ignite-metadata-update-" + streamerCfg.getKafkaPartsFrom() + "-" + streamerCfg.getKafkaPartsTo()));
+
+ cnsmr = new KafkaConsumer<>(kafkaProps);
+
+ cnsmr.subscribe(Collections.singletonList(streamerCfg.getMetadataTopic()));
+ }
+
+ /** {@inheritDoc} */
+ @Override public void run() {
+ U.setCurrentIgniteName(ign.name());
+
+ while (!stopped.get()) {
+ updateMetadata();
+
+ try {
+ Thread.sleep(metaUpdInterval);
+ }
+ catch (InterruptedException e) {
+ // Ignore.
+ }
+ }
+ }
+
+ /** Polls all available records from metadata topic and applies it to Ignite. */
+ public synchronized void updateMetadata() {
+ while (true) {
+ ConsumerRecords<Void, byte[]> recs = cnsmr.poll(Duration.ofMillis(kafkaReqTimeout));
+
+ if (recs.count() == 0)
+ return;
+
+ if (log.isInfoEnabled())
+ log.info("Polled from meta topic [rcvdEvts=" + rcvdEvts.addAndGet(recs.count()) + ']');
+
+ for (ConsumerRecord<Void, byte[]> rec : recs) {
+ Object data = IgniteUtils.fromBytes(rec.value());
+
+ if (data instanceof BinaryMetadata)
+ CdcEventsApplier.registerBinaryMeta(ign, log, (BinaryMetadata)data);
+ else if (data instanceof TypeMapping)
+ CdcEventsApplier.registerMapping(ign, log, (TypeMapping)data);
+ else
+ throw new IllegalArgumentException("Unknown meta type[type=" + data + ']');
+ }
+
+ cnsmr.commitSync(Duration.ofMillis(kafkaReqTimeout));
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void close() {
+ log.warning("Close applier!");
+
+ cnsmr.wakeup();
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(KafkaToIgniteMetadataUpdater.class, this);
+ }
+}
diff --git a/modules/cdc-ext/src/test/java/org/apache/ignite/cdc/AbstractReplicationTest.java b/modules/cdc-ext/src/test/java/org/apache/ignite/cdc/AbstractReplicationTest.java
index 4972267..ddedbc8 100644
--- a/modules/cdc-ext/src/test/java/org/apache/ignite/cdc/AbstractReplicationTest.java
+++ b/modules/cdc-ext/src/test/java/org/apache/ignite/cdc/AbstractReplicationTest.java
@@ -255,9 +255,6 @@ public abstract class AbstractReplicationTest extends GridCommonAbstractTest {
try {
IgniteCache<Integer, ConflictResolvableTestData> destCache = createCache(destCluster[0], ACTIVE_PASSIVE_CACHE);
- destCache.put(KEYS_CNT + 1, ConflictResolvableTestData.create());
- destCache.remove(KEYS_CNT + 1);
-
// Updates for "ignored-cache" should be ignored because of CDC consume configuration.
runAsync(generateData(IGNORED_CACHE, srcCluster[srcCluster.length - 1], IntStream.range(0, KEYS_CNT)));
runAsync(generateData(ACTIVE_PASSIVE_CACHE, srcCluster[srcCluster.length - 1], IntStream.range(0, KEYS_CNT)));
@@ -320,9 +317,6 @@ public abstract class AbstractReplicationTest extends GridCommonAbstractTest {
executeSql(srcCluster[0], createTbl);
executeSql(destCluster[0], createTbl);
- addData.accept(destCluster[0], -1);
- executeSql(destCluster[0], "DELETE FROM " + name);
-
IntStream.range(0, KEYS_CNT).forEach(i -> addData.accept(srcCluster[0], i));
List<IgniteInternalFuture<?>> futs = startActivePassiveCdc(name);
@@ -369,9 +363,6 @@ public abstract class AbstractReplicationTest extends GridCommonAbstractTest {
executeSql(srcCluster[0], createTbl);
executeSql(destCluster[0], createTbl);
- executeSql(destCluster[0], insertQry, -1, "Name-1");
- executeSql(destCluster[0], deleteQry);
-
IntStream.range(0, KEYS_CNT).forEach(id -> executeSql(srcCluster[0], insertQry, id, "Name" + id));
List<IgniteInternalFuture<?>> futs = startActivePassiveCdc("T1");
diff --git a/modules/cdc-ext/src/test/java/org/apache/ignite/cdc/kafka/CdcKafkaReplicationAppsTest.java b/modules/cdc-ext/src/test/java/org/apache/ignite/cdc/kafka/CdcKafkaReplicationAppsTest.java
index 847cd9e..1d88321 100644
--- a/modules/cdc-ext/src/test/java/org/apache/ignite/cdc/kafka/CdcKafkaReplicationAppsTest.java
+++ b/modules/cdc-ext/src/test/java/org/apache/ignite/cdc/kafka/CdcKafkaReplicationAppsTest.java
@@ -50,6 +50,9 @@ public class CdcKafkaReplicationAppsTest extends CdcKafkaReplicationTest {
/** */
public static final String TOPIC = "TOPIC";
+ /** */
+ public static final String METADATA_TOPIC = "METADATA_TOPIC";
+
/** */
public static final String CONSISTENT_ID = "CONSISTENT_ID";
@@ -95,12 +98,18 @@ public class CdcKafkaReplicationAppsTest extends CdcKafkaReplicationTest {
}
/** {@inheritDoc} */
- @Override protected IgniteInternalFuture<?> igniteToKafka(IgniteConfiguration igniteCfg, String topic, String cache) {
+ @Override protected IgniteInternalFuture<?> igniteToKafka(
+ IgniteConfiguration igniteCfg,
+ String topic,
+ String metadataTopic,
+ String cache
+ ) {
Map<String, String> params = new HashMap<>();
params.put(INSTANCE_NAME, igniteCfg.getIgniteInstanceName());
params.put(REPLICATED_CACHE, cache);
params.put(TOPIC, topic);
+ params.put(METADATA_TOPIC, metadataTopic);
params.put(CONSISTENT_ID, String.valueOf(igniteCfg.getConsistentId()));
params.put(PARTS, Integer.toString(DFLT_PARTS));
params.put(MAX_BATCH_SIZE, Integer.toString(KEYS_CNT));
@@ -116,6 +125,7 @@ public class CdcKafkaReplicationAppsTest extends CdcKafkaReplicationTest {
@Override protected IgniteInternalFuture<?> kafkaToIgnite(
String cacheName,
String topic,
+ String metadataTopic,
IgniteConfiguration igniteCfg,
int partFrom,
int partTo
@@ -129,6 +139,7 @@ public class CdcKafkaReplicationAppsTest extends CdcKafkaReplicationTest {
params.put(DISCO_PORT_RANGE, Integer.toString(discoPort + DFLT_PORT_RANGE));
params.put(REPLICATED_CACHE, cacheName);
params.put(TOPIC, topic);
+ params.put(METADATA_TOPIC, metadataTopic);
params.put(PROPS_PATH, kafkaPropsPath);
params.put(PARTS_FROM, Integer.toString(partFrom));
params.put(PARTS_TO, Integer.toString(partTo));
diff --git a/modules/cdc-ext/src/test/java/org/apache/ignite/cdc/kafka/CdcKafkaReplicationTest.java b/modules/cdc-ext/src/test/java/org/apache/ignite/cdc/kafka/CdcKafkaReplicationTest.java
index 481ceb6..e3663ab 100644
--- a/modules/cdc-ext/src/test/java/org/apache/ignite/cdc/kafka/CdcKafkaReplicationTest.java
+++ b/modules/cdc-ext/src/test/java/org/apache/ignite/cdc/kafka/CdcKafkaReplicationTest.java
@@ -47,6 +47,12 @@ public class CdcKafkaReplicationTest extends AbstractReplicationTest {
/** */
public static final String DEST_SRC_TOPIC = "dest-source";
+ /** */
+ public static final String SRC_DEST_META_TOPIC = "source-dest-meta";
+
+ /** */
+ public static final String DEST_SRC_META_TOPIC = "dest-source-meta";
+
/** */
public static final int DFLT_PARTS = 16;
@@ -65,6 +71,8 @@ public class CdcKafkaReplicationTest extends AbstractReplicationTest {
KAFKA.createTopic(SRC_DEST_TOPIC, DFLT_PARTS, 1);
KAFKA.createTopic(DEST_SRC_TOPIC, DFLT_PARTS, 1);
+ KAFKA.createTopic(SRC_DEST_META_TOPIC, 1, 1);
+ KAFKA.createTopic(DEST_SRC_META_TOPIC, 1, 1);
}
/** {@inheritDoc} */
@@ -97,12 +105,13 @@ public class CdcKafkaReplicationTest extends AbstractReplicationTest {
List<IgniteInternalFuture<?>> futs = new ArrayList<>();
for (IgniteEx ex : srcCluster)
- futs.add(igniteToKafka(ex.configuration(), cache, cache));
+ futs.add(igniteToKafka(ex.configuration(), cache, SRC_DEST_META_TOPIC, cache));
for (int i = 0; i < destCluster.length; i++) {
futs.add(kafkaToIgnite(
cache,
cache,
+ SRC_DEST_META_TOPIC,
destClusterCliCfg[i],
i * (DFLT_PARTS / 2),
(i + 1) * (DFLT_PARTS / 2)
@@ -117,13 +126,28 @@ public class CdcKafkaReplicationTest extends AbstractReplicationTest {
List<IgniteInternalFuture<?>> futs = new ArrayList<>();
for (IgniteEx ex : srcCluster)
- futs.add(igniteToKafka(ex.configuration(), SRC_DEST_TOPIC, ACTIVE_ACTIVE_CACHE));
+ futs.add(igniteToKafka(ex.configuration(), SRC_DEST_TOPIC, SRC_DEST_META_TOPIC, ACTIVE_ACTIVE_CACHE));
for (IgniteEx ex : destCluster)
- futs.add(igniteToKafka(ex.configuration(), DEST_SRC_TOPIC, ACTIVE_ACTIVE_CACHE));
-
- futs.add(kafkaToIgnite(ACTIVE_ACTIVE_CACHE, SRC_DEST_TOPIC, destClusterCliCfg[0], 0, DFLT_PARTS));
- futs.add(kafkaToIgnite(ACTIVE_ACTIVE_CACHE, DEST_SRC_TOPIC, srcClusterCliCfg[0], 0, DFLT_PARTS));
+ futs.add(igniteToKafka(ex.configuration(), DEST_SRC_TOPIC, DEST_SRC_META_TOPIC, ACTIVE_ACTIVE_CACHE));
+
+ futs.add(kafkaToIgnite(
+ ACTIVE_ACTIVE_CACHE,
+ SRC_DEST_TOPIC,
+ SRC_DEST_META_TOPIC,
+ destClusterCliCfg[0],
+ 0,
+ DFLT_PARTS
+ ));
+
+ futs.add(kafkaToIgnite(
+ ACTIVE_ACTIVE_CACHE,
+ DEST_SRC_TOPIC,
+ DEST_SRC_META_TOPIC,
+ srcClusterCliCfg[0],
+ 0,
+ DFLT_PARTS
+ ));
return futs;
}
@@ -138,13 +162,20 @@ public class CdcKafkaReplicationTest extends AbstractReplicationTest {
/**
* @param igniteCfg Ignite configuration.
* @param topic Kafka topic name.
+ * @param metadataTopic Metadata topic name.
* @param cache Cache name to stream to kafka.
* @return Future for Change Data Capture application.
*/
- protected IgniteInternalFuture<?> igniteToKafka(IgniteConfiguration igniteCfg, String topic, String cache) {
+ protected IgniteInternalFuture<?> igniteToKafka(
+ IgniteConfiguration igniteCfg,
+ String topic,
+ String metadataTopic,
+ String cache
+ ) {
return runAsync(() -> {
IgniteToKafkaCdcStreamer cdcCnsmr = new IgniteToKafkaCdcStreamer()
.setTopic(topic)
+ .setMetadataTopic(metadataTopic)
.setKafkaPartitions(DFLT_PARTS)
.setCaches(Collections.singleton(cache))
.setMaxBatchSize(KEYS_CNT)
@@ -173,6 +204,7 @@ public class CdcKafkaReplicationTest extends AbstractReplicationTest {
protected IgniteInternalFuture<?> kafkaToIgnite(
String cacheName,
String topic,
+ String metadataTopic,
IgniteConfiguration igniteCfg,
int fromPart,
int toPart
@@ -185,6 +217,7 @@ public class CdcKafkaReplicationTest extends AbstractReplicationTest {
cfg.setCaches(Collections.singletonList(cacheName));
cfg.setTopic(topic);
+ cfg.setMetadataTopic(metadataTopic);
cfg.setKafkaRequestTimeout(DFLT_KAFKA_REQ_TIMEOUT);
return runAsync(new KafkaToIgniteCdcStreamer(igniteCfg, kafkaProperties(), cfg));
diff --git a/modules/cdc-ext/src/test/resources/loader/kafka-to-ignite-correct.xml b/modules/cdc-ext/src/test/resources/loader/kafka-to-ignite-correct.xml
index 016eafa..78a8809 100644
--- a/modules/cdc-ext/src/test/resources/loader/kafka-to-ignite-correct.xml
+++ b/modules/cdc-ext/src/test/resources/loader/kafka-to-ignite-correct.xml
@@ -38,6 +38,7 @@
<bean id="streamer.cfg" class="org.apache.ignite.cdc.kafka.KafkaToIgniteCdcStreamerConfiguration">
<property name="topic" value="ignite" />
+ <property name="metadataTopic" value="ignite-metadata" />
<property name="kafkaPartsFrom" value="0" />
<property name="kafkaPartsTo" value="16" />
</bean>
diff --git a/modules/cdc-ext/src/test/resources/replication/ignite-to-kafka.xml b/modules/cdc-ext/src/test/resources/replication/ignite-to-kafka.xml
index 2f72058..75ca237 100644
--- a/modules/cdc-ext/src/test/resources/replication/ignite-to-kafka.xml
+++ b/modules/cdc-ext/src/test/resources/replication/ignite-to-kafka.xml
@@ -45,6 +45,7 @@
<property name="consumer">
<bean class="org.apache.ignite.cdc.kafka.IgniteToKafkaCdcStreamer">
<property name="topic" value="{TOPIC}" />
+ <property name="metadataTopic" value="{METADATA_TOPIC}" />
<property name="kafkaPartitions" value="{PARTS}" />
<property name="caches">
<util:list>
diff --git a/modules/cdc-ext/src/test/resources/replication/kafka-to-ignite.xml b/modules/cdc-ext/src/test/resources/replication/kafka-to-ignite.xml
index 8bb35a4..3a59f88 100644
--- a/modules/cdc-ext/src/test/resources/replication/kafka-to-ignite.xml
+++ b/modules/cdc-ext/src/test/resources/replication/kafka-to-ignite.xml
@@ -52,6 +52,7 @@
</util:list>
</property>
<property name="topic" value="{TOPIC}" />
+ <property name="metadataTopic" value="{METADATA_TOPIC}" />
<property name="kafkaPartsFrom" value="{PARTS_FROM}"/>
<property name="kafkaPartsTo" value="{PARTS_TO}"/>
<property name="threadCount" value="{THREAD_CNT}"/>