Bump libthrift from 0.9.3 to 0.12.0

hbase-common/src/main/java/org/apache/hadoop/hbase/util/KafkaUtil.java

@@ -17,6 +17,9 @@
  * limitations under the License.
  */
 package org.apache.hadoop.hbase.util;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.hbase.classification.InterfaceAudience;
@@ -28,23 +31,51 @@ public class KafkaUtil {
   private static final Log LOG = LogFactory.getLog(KafkaUtil.class);
   public static final String KAFKA_BROKER_SERVERS = "kafka.bootstrap.servers";
 
+  private static final String LOCAL_ZONE;
+  static {
+    String hostname = null;
+    try {
+      hostname = InetAddress.getLocalHost().getHostName();
+    } catch (UnknownHostException e) {
+      LOG.warn("could not determin local hostname.", e);
+    }
+    int index = hostname == null ? -1 : hostname.indexOf("-");
+    LOCAL_ZONE = index == -1 ? "rz" : hostname.substring(0, index);
+  }
+
   /**
    * Each HBase table corresponding to an KAFKA topic
    */
   public static String getTableTopic(String tableName) {
     return tableName;
   }
 
+  public static String getConsumerGroup(String tableName) {
+    return new StringBuilder("connect-").append(tableName).append("-").append(LOCAL_ZONE).toString();
+  }
+
   public static String getTopicTable(String topicName) {
     return topicName;
   }
 
   /**
+   * return the partition this row will go to
    * Each HBase Region corresponding to an KAFKA partition,
    * rowkey should contains partition info, split by '_'
    */
   public static int getTablePartition(String rowkey, int partitionCount) {
-    return getPrefix(rowkey) % partitionCount;
+    String cntStr = String.valueOf(partitionCount);
+    int len = cntStr.length();
+    assert(len <= 5); // partitionCount's upper limit is 10000, len is 5
+    int interval = (partitionCount == Math.pow(10, len-1)) ? 1 : (int) Math.pow(10, len) / partitionCount;
+    int prefix = getPrefix(rowkey);
+    if (interval == 1) { // power of 10
+      return prefix / (int) Math.pow(10, 5 - len);
+    } else {
+      String preStr = String.format("%04d", prefix).substring(0, len);
+      int value = Integer.parseInt(preStr) / interval;
+      return value;
+    }
   }
 
   public static int getPrefix(String rowkey) {
@@ -54,7 +85,10 @@ public static int getPrefix(String rowkey) {
     try {
       int index = rowkey.indexOf("_");
       String prefix = (index == -1) ? rowkey : rowkey.substring(0, index);
-      return Integer.parseInt(prefix);
+      for (int i = prefix.length(); i < 4; i++) { // rowkey has 4 prefix
+        prefix += "0";
+      }
+      return Integer.parseInt(prefix.substring(0, 4)); // only retain 4 num
     } catch (NumberFormatException e) {
       LOG.warn(e.getMessage(), e);
       return 0;

hbase-common/src/test/java/org/apache/hadoop/hbase/util/TestKafkaUtil.java

@@ -0,0 +1,73 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.util;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+@Category(SmallTests.class)
+public class TestKafkaUtil {
+  @Test
+  public void testGetTablePartition() {
+    assertEquals(1, KafkaUtil.getTablePartition("1247_rowkey", 10));
+    assertEquals(12, KafkaUtil.getTablePartition("1247_rowkey", 100));
+    assertEquals(124, KafkaUtil.getTablePartition("1247_rowkey", 1000));
+    assertEquals(1247, KafkaUtil.getTablePartition("1247_rowkey", 10000));
+
+    // partition count is 100
+    for (int i = 0; i < 10000; i++) {
+      String rowkey = String.format("%04d", i);
+      int partition = KafkaUtil.getTablePartition(rowkey, 100);
+      if (partition > 9) {
+        assertTrue(rowkey.startsWith(String.valueOf(partition)));
+      } else {
+        assertTrue(rowkey.startsWith("0" + String.valueOf(partition)));
+      }
+    }
+
+    // partition count is 200
+    Map<Integer, Integer> partInfo = new HashMap<>();
+    int interval = 10000 / 200;
+    for (int i = 0; i < 200; i++) {
+      partInfo.put(i, i * interval);
+    }
+    partInfo.put(200, Integer.MAX_VALUE);
+    for (int i = 0; i < 10000; i++) {
+      String rowkey = String.format("%04d", i);
+      int partition = KafkaUtil.getTablePartition(rowkey, 200);
+      assertTrue(partInfo.get(partition) <= i && i < partInfo.get(partition + 1));
+    }
+  }
+
+  @Test
+  public void testGetConsumerGroup() throws UnknownHostException {
+    String table = "usertable";
+    String hostname = InetAddress.getLocalHost().getHostName();
+    String groupName = "connect-" + table + "-" + hostname.substring(0, hostname.indexOf("-"));
+    assertEquals(groupName, KafkaUtil.getConsumerGroup("usertable"));
+  }
+}

hbase-server/src/main/java/org/apache/hadoop/hbase/SplitLogTask.java

@@ -35,10 +35,15 @@
  */
 @InterfaceAudience.Private
 public class SplitLogTask {
+  public static final int KAFKA_TASK_FIELD_LEN = 5;
   private final ServerName originServer;
   private final ZooKeeperProtos.SplitLogTask.State state;
   private final ZooKeeperProtos.SplitLogTask.RecoveryMode mode;
 
+  public enum Type {
+    HDFS, KAFKA;
+  }
+
   public static class Unassigned extends SplitLogTask {
     public Unassigned(final ServerName originServer, final RecoveryMode mode) {
       super(originServer, ZooKeeperProtos.SplitLogTask.State.UNASSIGNED, mode);
@@ -168,6 +173,14 @@ public static SplitLogTask parseFrom(final byte [] data) throws DeserializationE
     }
   }
 
+  public static Type getTaskType(String taskName) {
+    if (taskName.split("_").length == KAFKA_TASK_FIELD_LEN) {
+      return Type.KAFKA;
+    } else {
+      return Type.HDFS;
+    }
+  }
+
   /**
    * @return This instance serialized into a byte array
    * @see #parseFrom(byte[])

hbase-server/src/main/java/org/apache/hadoop/hbase/coordination/SplitLogManagerCoordination.java

@@ -21,11 +21,13 @@
 
 import java.io.IOException;
 import java.io.InterruptedIOException;
+import java.util.Collection;
 import java.util.Set;
 import java.util.concurrent.ConcurrentMap;
 
 import org.apache.hadoop.hbase.HRegionInfo;
 import org.apache.hadoop.hbase.ServerName;
+import org.apache.hadoop.hbase.SplitLogTask;
 import org.apache.hadoop.hbase.classification.InterfaceAudience;
 import org.apache.hadoop.hbase.coordination.ZKSplitLogManagerCoordination.TaskFinisher;
 import org.apache.hadoop.hbase.master.LogRecoveryManager.ResubmitDirective;
@@ -105,12 +107,12 @@ public ServerName getServerName() {
   /**
    * Provide the configuration from the SplitLogManager
    */
-  void setDetails(SplitLogManagerDetails details);
+  void addDetails(SplitLogTask.Type type, SplitLogManagerDetails details);
 
   /**
    * Returns the configuration that was provided previously
    */
-  SplitLogManagerDetails getDetails();
+  Collection<SplitLogManagerDetails> getDetails();
 
   /**
    * Prepare the new task
@@ -211,5 +213,5 @@ void removeStaleRecoveringRegions(Set<String> knownServers) throws IOException,
   @VisibleForTesting
   void init() throws IOException;
 
-  public void setTaskFinisher(TaskFinisher taskFinisher);
+  public void addTaskFinisher(TaskFinisher taskFinisher);
 }