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 compiler

Browse Source
This commit is contained in:
agapple
2019-03-28 10:23:33 +08:00
parent 20e7becfe9
commit 97640b717b
1 changed files with 12 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java
View File

@@ -139,7 +139,7 @@ public class CanalKafkaProducer implements CanalMQProducer {
private void send(MQProperties.CanalDestination canalDestination, String topicName, Message message) private void send(MQProperties.CanalDestination canalDestination, String topicName, Message message)
throws Exception { throws Exception {
if (!kafkaProperties.getFlatMessage()) { if (!kafkaProperties.getFlatMessage()) {
List<ProducerRecord<String, Message>> records = new ArrayList<>(); List<ProducerRecord<String, Message>> records = new ArrayList<ProducerRecord<String, Message>>();
if (canalDestination.getPartitionHash() != null && !canalDestination.getPartitionHash().isEmpty()) { if (canalDestination.getPartitionHash() != null && !canalDestination.getPartitionHash().isEmpty()) {
Message[] messages = MQMessageUtils.messagePartition(message, Message[] messages = MQMessageUtils.messagePartition(message,
canalDestination.getPartitionsNum(), canalDestination.getPartitionsNum(),
@@ -148,12 +148,12 @@ public class CanalKafkaProducer implements CanalMQProducer {
for (int i = 0; i < length; i++) { for (int i = 0; i < length; i++) {
Message messagePartition = messages[i]; Message messagePartition = messages[i];
if (messagePartition != null) { if (messagePartition != null) {
records.add(new ProducerRecord<>(topicName, i, null, messagePartition)); records.add(new ProducerRecord<String, Message>(topicName, i, null, messagePartition));
} }
} }
} else { } else {
final int partition = canalDestination.getPartition() != null ? canalDestination.getPartition() : 0; final int partition = canalDestination.getPartition() != null ? canalDestination.getPartition() : 0;
records.add(new ProducerRecord<>(topicName, partition, null, message)); records.add(new ProducerRecord<String, Message>(topicName, partition, null, message));
} }
if (!records.isEmpty()) { if (!records.isEmpty()) {