chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

fixed issue #3829 , kafka connector support k8s env

Browse Source
This commit is contained in:
jianghang.loujh
2021-10-09 16:31:59 +08:00
parent 0a2254eeaf
commit f26b84ffc2
2 changed files with 5 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
connector/kafka-connector/src/main/java/com/alibaba/otter/canal/connector/kafka/consumer/CanalKafkaConsumer.java
View File

@@ -9,6 +9,7 @@ import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import com.alibaba.otter.canal.common.utils.PropertiesUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.clients.consumer.KafkaConsumer;
@@ -53,6 +54,8 @@ public class CanalKafkaConsumer implements CanalMsgConsumer {
String k = (String) entry.getKey(); String k = (String) entry.getKey();
Object v = entry.getValue(); Object v = entry.getValue();
if (k.startsWith(PREFIX_KAFKA_CONFIG) && v != null) { if (k.startsWith(PREFIX_KAFKA_CONFIG) && v != null) {
// check env config
v = PropertiesUtils.getProperty(properties, k);
kafkaProperties.put(k.substring(PREFIX_KAFKA_CONFIG.length()), v); kafkaProperties.put(k.substring(PREFIX_KAFKA_CONFIG.length()), v);
} }
} }

2
connector/kafka-connector/src/main/java/com/alibaba/otter/canal/connector/kafka/producer/CanalKafkaProducer.java
View File

@@ -102,6 +102,8 @@ public class CanalKafkaProducer extends AbstractMQProducer implements CanalMQPro
String key = (String) entry.getKey(); String key = (String) entry.getKey();
Object value = entry.getValue(); Object value = entry.getValue();
if (key.startsWith(PREFIX_KAFKA_CONFIG) && value != null) { if (key.startsWith(PREFIX_KAFKA_CONFIG) && value != null) {
// check env config
value = PropertiesUtils.getProperty(properties, key);
key = key.substring(PREFIX_KAFKA_CONFIG.length()); key = key.substring(PREFIX_KAFKA_CONFIG.length());
kafkaProperties.put(key, value); kafkaProperties.put(key, value);
} }