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 da1a874c IGNITE-19369 Update metadata topic offset only after commit
(#212)
da1a874c is described below
commit da1a874c27b93923f07d9c0a6876dcca1146a9c2
Author: Nikolay <[email protected]>
AuthorDate: Wed Apr 26 15:56:43 2023 +0300
IGNITE-19369 Update metadata topic offset only after commit (#212)
---
.../ignite/cdc/kafka/IgniteToKafkaCdcStreamer.java | 4 +-
.../cdc/kafka/KafkaToIgniteMetadataUpdater.java | 44 +++++++++++++++++++---
2 files changed, 40 insertions(+), 8 deletions(-)
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 a7117480..0dce9442 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
@@ -201,7 +201,7 @@ public class IgniteToKafkaCdcStreamer implements
CdcConsumer {
@Override public void onTypes(Iterator<BinaryType> types) {
sendAll(
types,
- t -> new ProducerRecord<>(metadataTopic,
IgniteUtils.toBytes(((BinaryTypeImpl)t).metadata())),
+ t -> new ProducerRecord<>(metadataTopic, 0, null,
IgniteUtils.toBytes(((BinaryTypeImpl)t).metadata())),
typesCnt
);
@@ -212,7 +212,7 @@ public class IgniteToKafkaCdcStreamer implements
CdcConsumer {
@Override public void onMappings(Iterator<TypeMapping> mappings) {
sendAll(
mappings,
- m -> new ProducerRecord<>(metadataTopic, IgniteUtils.toBytes(m)),
+ m -> new ProducerRecord<>(metadataTopic, 0, null,
IgniteUtils.toBytes(m)),
mappingsCnt
);
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
index 15e793aa..b8e8a9ad 100644
---
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
@@ -25,6 +25,7 @@ import java.util.Properties;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
+import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.cdc.TypeMapping;
import org.apache.ignite.internal.binary.BinaryContext;
@@ -35,6 +36,8 @@ import org.apache.ignite.internal.util.typedef.internal.S;
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.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.consumer.OffsetCommitCallback;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.VoidDeserializer;
@@ -46,7 +49,7 @@ import static
org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_
import static
org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG;
/** */
-public class KafkaToIgniteMetadataUpdater implements AutoCloseable {
+public class KafkaToIgniteMetadataUpdater implements AutoCloseable,
OffsetCommitCallback {
/** Binary context. */
private final BinaryContext ctx;
@@ -63,7 +66,10 @@ public class KafkaToIgniteMetadataUpdater implements
AutoCloseable {
private final AtomicLong rcvdEvts = new AtomicLong();
/** Offsets from the last successful metadata update. */
- private Map<TopicPartition, Long> offsets;
+ private volatile Map<TopicPartition, Long> offsets;
+
+ /** Possible commit error. */
+ private volatile Exception err;
/** Metadata topic partitions. */
private final Set<TopicPartition> parts;
@@ -112,6 +118,9 @@ public class KafkaToIgniteMetadataUpdater implements
AutoCloseable {
/** Polls all available records from metadata topic and applies it to
Ignite. */
public synchronized void updateMetadata() {
+ if (err != null)
+ throw new IgniteException(err);
+
// If there are no new records in topic, method KafkaConsumer#poll
blocks up to the specified timeout.
// In order to eliminate this, we compare current offsets with the
offsets from the last metadata update
// (stored in 'offsets' field). If there are no offsets changes,
polling cycle is skipped.
@@ -124,13 +133,15 @@ public class KafkaToIgniteMetadataUpdater implements
AutoCloseable {
return;
}
- offsets = new HashMap<>(offsets0);
-
while (true) {
ConsumerRecords<Void, byte[]> recs =
cnsmr.poll(Duration.ofMillis(kafkaReqTimeout));
- if (recs.count() == 0)
+ if (recs.count() == 0) {
+ if (log.isDebugEnabled())
+ log.debug("Empty poll from meta topic");
+
return;
+ }
if (log.isInfoEnabled())
log.info("Polled from meta topic [rcvdEvts=" +
rcvdEvts.addAndGet(recs.count()) + ']');
@@ -146,8 +157,29 @@ public class KafkaToIgniteMetadataUpdater implements
AutoCloseable {
throw new IllegalArgumentException("Unknown meta
type[type=" + data + ']');
}
- cnsmr.commitSync(Duration.ofMillis(kafkaReqTimeout));
+ // Offsets updated only after commit.
+ cnsmr.commitAsync(this);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onComplete(Map<TopicPartition, OffsetAndMetadata>
committed, Exception err) {
+ if (err != null) {
+ log.warning("Commit error:", err);
+
+ this.err = err;
+
+ return;
}
+
+ Map<TopicPartition, Long> offsets0 = new HashMap<>();
+
+ committed.forEach((tp, offAndMeta) -> offsets0.put(tp,
offAndMeta.offset()));
+
+ if (log.isDebugEnabled())
+ log.debug("Offset committed: " + offsets0);
+
+ offsets = offsets0;
}
/** {@inheritDoc} */
