You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by GitBox <gi...@apache.org> on 2022/02/22 22:13:05 UTC
[GitHub] [ignite-extensions] ololo3000 commented on a change in pull request #99: IGNITE-16176 Adds configurable Kafka requests timeout.
ololo3000 commented on a change in pull request #99:
URL: https://github.com/apache/ignite-extensions/pull/99#discussion_r812403505
##########
File path: modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/IgniteToKafkaCdcStreamer.java
##########
@@ -84,22 +98,28 @@
private KafkaProducer<Integer, byte[]> producer;
/** Handle only primary entry flag. */
- private final boolean onlyPrimary;
+ private boolean onlyPrimary = DFLT_IS_ONLY_PRIMARY;
/** Topic name. */
- private final String topic;
+ private String topic = DFLT_TOPIC;
Review comment:
Done. I have done the symmetric changes for KafkaToIgniteCdcStreamer too.
##########
File path: modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/IgniteToKafkaCdcStreamer.java
##########
@@ -84,22 +98,28 @@
private KafkaProducer<Integer, byte[]> producer;
/** Handle only primary entry flag. */
- private final boolean onlyPrimary;
+ private boolean onlyPrimary = DFLT_IS_ONLY_PRIMARY;
/** Topic name. */
- private final String topic;
+ private String topic = DFLT_TOPIC;
/** Kafka topic partitions count. */
- private final int kafkaParts;
+ private int kafkaParts = DFLT_PARTS;
/** Kafka properties. */
- private final Properties kafkaProps;
+ private Properties kafkaProps;
/** Cache IDs. */
- private final Set<Integer> cachesIds;
+ private Set<Integer> cachesIds;
+
+ /** Cache names. */
+ private Collection<String> cacheNames = DFLT_CACHES;
Review comment:
Done.
##########
File path: modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/IgniteToKafkaCdcStreamer.java
##########
@@ -110,37 +130,17 @@
/** Count of sent messages. */
private AtomicLongMetric msgsSnt;
+ /** */
+ public IgniteToKafkaCdcStreamer() {
+ // No-op.
+ }
+
/**
- * @param topic Topic name.
- * @param kafkaParts Kafka partitions count.
- * @param caches Cache names.
- * @param maxBatchSize Maximum size of records concurrently sent to Kafka.
- * @param onlyPrimary If {@code true} then stream only events from primaries.
* @param kafkaProps Kafka properties.
*/
- public IgniteToKafkaCdcStreamer(
- String topic,
- int kafkaParts,
- Set<String> caches,
- int maxBatchSize,
- boolean onlyPrimary,
- Properties kafkaProps
- ) {
- assert caches != null && !caches.isEmpty();
-
- this.topic = topic;
- this.kafkaParts = kafkaParts;
- this.onlyPrimary = onlyPrimary;
+ public IgniteToKafkaCdcStreamer(Properties kafkaProps) {
Review comment:
Yes. Removed.
##########
File path: modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/IgniteToKafkaCdcStreamer.java
##########
@@ -84,22 +98,28 @@
private KafkaProducer<Integer, byte[]> producer;
/** Handle only primary entry flag. */
- private final boolean onlyPrimary;
+ private boolean onlyPrimary = DFLT_IS_ONLY_PRIMARY;
/** Topic name. */
- private final String topic;
+ private String topic = DFLT_TOPIC;
/** Kafka topic partitions count. */
- private final int kafkaParts;
+ private int kafkaParts = DFLT_PARTS;
/** Kafka properties. */
- private final Properties kafkaProps;
+ private Properties kafkaProps;
/** Cache IDs. */
- private final Set<Integer> cachesIds;
+ private Set<Integer> cachesIds;
+
+ /** Cache names. */
+ private Collection<String> cacheNames = DFLT_CACHES;
/** Max batch size. */
- private final int maxBatchSize;
+ private int maxBatchSize = DFLT_MAX_BATCH_SIZE;
+
+ /** The maximum time to complete Kafka related requests, in seconds. */
+ private int kafkaReqTimeout = DFLT_REQ_TIMEOUT;
Review comment:
Done.
##########
File path: modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/IgniteToKafkaCdcStreamer.java
##########
@@ -194,7 +194,7 @@ public IgniteToKafkaCdcStreamer(
if (!futs.isEmpty()) {
try {
for (Future<RecordMetadata> fut : futs)
- fut.get(DFLT_REQ_TIMEOUT, TimeUnit.SECONDS);
+ fut.get(kafkaReqTimeout, TimeUnit.SECONDS);
Review comment:
Done.
##########
File path: modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/IgniteToKafkaCdcStreamer.java
##########
@@ -213,6 +213,16 @@ public IgniteToKafkaCdcStreamer(
/** {@inheritDoc} */
@Override public void start(MetricRegistry mreg) {
+ if (kafkaProps == null)
Review comment:
Done.
##########
File path: modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/IgniteToKafkaCdcStreamer.java
##########
@@ -213,6 +213,16 @@ public IgniteToKafkaCdcStreamer(
/** {@inheritDoc} */
@Override public void start(MetricRegistry mreg) {
+ if (kafkaProps == null)
+ throw new IllegalStateException("The Kafka properties must be set in order to initiate a connection.");
+
+ kafkaProps.setProperty(KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());
+ kafkaProps.setProperty(VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+
+ cachesIds = cacheNames.stream()
Review comment:
It seems that we can check it right in the setter. Done.
##########
File path: modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteCdcStreamerApplier.java
##########
@@ -114,6 +111,9 @@
/** */
private final AtomicLong rcvdEvts = new AtomicLong();
+ /** The maximum time to complete Kafka related requests, in seconds. */
+ private final int kafkaReqTimeout;
Review comment:
Done.
##########
File path: modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteCdcStreamerApplier.java
##########
@@ -211,7 +213,7 @@ private void poll(KafkaConsumer<Integer, byte[]> cnsmr) throws IgniteCheckedExce
apply(F.iterator(recs, this::deserialize, true, rec -> F.isEmpty(caches) || caches.contains(rec.key())));
- cnsmr.commitSync(Duration.ofSeconds(DFLT_REQ_TIMEOUT));
+ cnsmr.commitSync(Duration.ofSeconds(kafkaReqTimeout));
Review comment:
Done.
##########
File path: modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteCdcStreamerApplier.java
##########
@@ -201,7 +203,7 @@ public KafkaToIgniteCdcStreamerApplier(
* @param cnsmr Data consumer.
*/
private void poll(KafkaConsumer<Integer, byte[]> cnsmr) throws IgniteCheckedException {
- ConsumerRecords<Integer, byte[]> recs = cnsmr.poll(Duration.ofSeconds(DFLT_REQ_TIMEOUT));
+ ConsumerRecords<Integer, byte[]> recs = cnsmr.poll(Duration.ofSeconds(kafkaReqTimeout));
Review comment:
Done.
##########
File path: modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteCdcStreamerApplier.java
##########
@@ -182,7 +184,7 @@ public KafkaToIgniteCdcStreamerApplier(
finally {
for (KafkaConsumer<Integer, byte[]> consumer : cnsmrs) {
try {
- consumer.close(Duration.ofSeconds(DFLT_REQ_TIMEOUT));
+ consumer.close(Duration.ofSeconds(kafkaReqTimeout));
Review comment:
Done.
##########
File path: modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/IgniteToKafkaCdcStreamer.java
##########
@@ -213,6 +213,16 @@ public IgniteToKafkaCdcStreamer(
/** {@inheritDoc} */
@Override public void start(MetricRegistry mreg) {
+ if (kafkaProps == null)
Review comment:
Done.
I have done the symmetric check for KafkaToIgniteStreamer too.
##########
File path: modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/IgniteToKafkaCdcStreamer.java
##########
@@ -232,4 +242,139 @@ public IgniteToKafkaCdcStreamer(
@Override public void stop() {
producer.close();
}
+
+ /** @return Whether entries only from primary nodes should be handled. */
+ public boolean isOnlyPrimary() {
+ return onlyPrimary;
+ }
+
+ /**
+ * Sets whether entries only from primary nodes should be handled.
+ *
+ * @param onlyPrimary Kafka whether entries only from primary nodes should be handled.
+ * @return {@code this} for chaining.
+ */
+ public IgniteToKafkaCdcStreamer setOnlyPrimary(boolean onlyPrimary) {
+ assert producer == null;
Review comment:
Let's just remove this checks then.
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org