From 82ed9db8aea313421b7d6fa42bc10a84c38d8a65 Mon Sep 17 00:00:00 2001
From: agapple <jianghang115@gmail.com>
Date: Wed, 3 Apr 2019 11:26:31 +0800
Subject: [PATCH] fixed rocketmq partition out of range

---
 .../canal/rocketmq/CanalRocketMQProducer.java | 26 ++++++++++++-------
 1 file changed, 16 insertions(+), 10 deletions(-)

diff --git a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java
index fa1a2819..aaa658e2 100644
--- a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java
+++ b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java
@@ -107,11 +107,10 @@ public class CanalRocketMQProducer implements CanalMQProducer {
                                     @Override
                                     public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                                         if (index > mqs.size()) {
-                                            throw new CanalServerException("partition number is error,config num:"
-                                                                           + destination.getPartitionsNum()
-                                                                           + ", mq num: " + mqs.size());
+                                            return mqs.get(index % mqs.size());
+                                        } else {
+                                            return mqs.get(index);
                                         }
-                                        return mqs.get(index);
                                     }
                                 }, null);
                             } catch (Exception e) {
@@ -134,7 +133,11 @@ public class CanalRocketMQProducer implements CanalMQProducer {
 
                         @Override
                         public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
-                            return mqs.get(partition);
+                            if (partition > mqs.size()) {
+                                return mqs.get(partition % mqs.size());
+                            } else {
+                                return mqs.get(partition);
+                            }
                         }
                     }, null);
                 }
@@ -168,11 +171,10 @@ public class CanalRocketMQProducer implements CanalMQProducer {
                                         @Override
                                         public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                                             if (index > mqs.size()) {
-                                                throw new CanalServerException("partition number is error,config num:"
-                                                                               + destination.getPartitionsNum()
-                                                                               + ", mq num: " + mqs.size());
+                                                return mqs.get(index % mqs.size());
+                                            } else {
+                                                return mqs.get(index);
                                             }
-                                            return mqs.get(index);
                                         }
                                     }, null);
                                 } catch (Exception e) {
@@ -196,7 +198,11 @@ public class CanalRocketMQProducer implements CanalMQProducer {
 
                                 @Override
                                 public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
-                                    return mqs.get(partition);
+                                    if (partition > mqs.size()) {
+                                        return mqs.get(partition % mqs.size());
+                                    } else {
+                                        return mqs.get(partition);
+                                    }
                                 }
                             }, null);
                         } catch (Exception e) {