Don't close connections,TODO: concurrent issues

killme2008 · killme2008 · commit cbfc53477dc9 · 2013-07-25T00:57:55.000+08:00
diff --git a/metamorphosis-client/src/main/java/com/taobao/metamorphosis/client/consumer/ConsumerZooKeeper.java b/metamorphosis-client/src/main/java/com/taobao/metamorphosis/client/consumer/ConsumerZooKeeper.java
@@ -702,6 +702,7 @@ boolean rebalance() throws InterruptedException, Exception {
                     this.fetchManager.stopFetchRunner();
                     // closed all connections to old brokers.
                     this.closeOldBrokersConnections();
+                    this.commitOffsets();
                     this.updateFetchRunner(cluster);
                     this.oldCluster = cluster;
                 }
@@ -789,7 +790,8 @@ boolean rebalance() throws InterruptedException, Exception {
 
         private void closeOldBrokersConnections() throws NotifyRemotingException {
             for (Broker old : this.oldBrokerSet) {
-                ConsumerZooKeeper.this.remotingClient.closeWithRef(old.getZKString(), this, false);
+                // ConsumerZooKeeper.this.remotingClient.closeWithRef(old.getZKString(),
+                // this, false);
                 log.warn("Closed " + old.getZKString());
             }
         }

Original file line number	Diff line number	Diff line change
`@@ -702,6 +702,7 @@ boolean rebalance() throws InterruptedException, Exception {`
`702`	`702`	`this.fetchManager.stopFetchRunner();`
`703`	`703`	`// closed all connections to old brokers.`
`704`	`704`	`this.closeOldBrokersConnections();`
	`705`	`+ this.commitOffsets();`
`705`	`706`	`this.updateFetchRunner(cluster);`
`706`	`707`	`this.oldCluster = cluster;`
`707`	`708`	`}`
`@@ -789,7 +790,8 @@ boolean rebalance() throws InterruptedException, Exception {`
`789`	`790`
`790`	`791`	`private void closeOldBrokersConnections() throws NotifyRemotingException {`
`791`	`792`	`for (Broker old : this.oldBrokerSet) {`
`792`		`- ConsumerZooKeeper.this.remotingClient.closeWithRef(old.getZKString(), this, false);`
	`793`	`+ // ConsumerZooKeeper.this.remotingClient.closeWithRef(old.getZKString(),`
	`794`	`+ // this, false);`
`793`	`795`	`log.warn("Closed " + old.getZKString());`
`794`	`796`	`}`
`795`	`797`	`}`