From 97640b717b855ab8b56e3ca37026732e5d56eebd Mon Sep 17 00:00:00 2001 From: agapple Date: Thu, 28 Mar 2019 10:23:33 +0800 Subject: [PATCH] fixed compiler --- .../otter/canal/kafka/CanalKafkaProducer.java | 24 +++++++++---------- 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index c4605067..6cc6724f 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -139,7 +139,7 @@ public class CanalKafkaProducer implements CanalMQProducer { private void send(MQProperties.CanalDestination canalDestination, String topicName, Message message) throws Exception { if (!kafkaProperties.getFlatMessage()) { - List> records = new ArrayList<>(); + List> records = new ArrayList>(); if (canalDestination.getPartitionHash() != null && !canalDestination.getPartitionHash().isEmpty()) { Message[] messages = MQMessageUtils.messagePartition(message, canalDestination.getPartitionsNum(), @@ -148,23 +148,23 @@ public class CanalKafkaProducer implements CanalMQProducer { for (int i = 0; i < length; i++) { Message messagePartition = messages[i]; if (messagePartition != null) { - records.add(new ProducerRecord<>(topicName, i, null, messagePartition)); + records.add(new ProducerRecord(topicName, i, null, messagePartition)); } } } else { final int partition = canalDestination.getPartition() != null ? canalDestination.getPartition() : 0; - records.add(new ProducerRecord<>(topicName, partition, null, message)); + records.add(new ProducerRecord(topicName, partition, null, message)); } - if (!records.isEmpty()) { - for (ProducerRecord record : records) { - producer.send(record).get(); - } - - if (logger.isDebugEnabled()) { - logger.debug("Send message to kafka topic: [{}], packet: {}", topicName, message.toString()); - } - } + if (!records.isEmpty()) { + for (ProducerRecord record : records) { + producer.send(record).get(); + } + + if (logger.isDebugEnabled()) { + logger.debug("Send message to kafka topic: [{}], packet: {}", topicName, message.toString()); + } + } } else { // 发送扁平数据json List flatMessages = MQMessageUtils.messageConverter(message);