| /** |
| * 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 |
| * <p> |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * <p> |
| * 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.kafka; |
| |
| import java.io.File; |
| import java.util.Arrays; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.UUID; |
| import java.util.stream.Collectors; |
| |
| import org.apache.commons.lang3.StringUtils; |
| import org.apache.curator.framework.CuratorFramework; |
| import org.apache.curator.framework.CuratorFrameworkFactory; |
| import org.apache.curator.retry.ExponentialBackoffRetry; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.hbase.HBaseConfiguration; |
| import org.apache.hadoop.hbase.HConstants; |
| import org.apache.hadoop.hbase.ReplicationPeerNotFoundException; |
| import org.apache.hadoop.hbase.client.Admin; |
| import org.apache.hadoop.hbase.client.Connection; |
| import org.apache.hadoop.hbase.client.ConnectionFactory; |
| import org.apache.hadoop.hbase.regionserver.HRegionServer; |
| import org.apache.hadoop.hbase.regionserver.HRegionServerCommandLine; |
| import org.apache.hadoop.hbase.replication.ReplicationPeerConfig; |
| import org.apache.hadoop.hbase.replication.ReplicationPeerConfigBuilder; |
| import org.apache.hadoop.hbase.replication.ReplicationPeerDescription; |
| import org.apache.hadoop.hbase.util.Bytes; |
| import org.apache.hadoop.hbase.util.VersionInfo; |
| import org.apache.hadoop.util.GenericOptionsParser; |
| import org.apache.yetus.audience.InterfaceAudience; |
| |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import org.apache.hbase.thirdparty.org.apache.commons.cli.BasicParser; |
| import org.apache.hbase.thirdparty.org.apache.commons.cli.CommandLine; |
| import org.apache.hbase.thirdparty.org.apache.commons.cli.HelpFormatter; |
| import org.apache.hbase.thirdparty.org.apache.commons.cli.Options; |
| import org.apache.hbase.thirdparty.org.apache.commons.cli.ParseException; |
| |
| |
| |
| /** |
| * hbase to kafka bridge. |
| * |
| * Starts up a region server and receives replication events, just like a peer |
| * cluster member. It takes the events and cell by cell determines how to |
| * route them (see kafka-route-rules.xml) |
| */ |
| @InterfaceAudience.Private |
| public final class KafkaProxy { |
| private static final Logger LOG = LoggerFactory.getLogger(KafkaProxy.class); |
| |
| public static final String KAFKA_PROXY_RULES_FILE = "kafkaproxy.rule.file"; |
| public static final String KAFKA_PROXY_KAFKA_PROPERTIES = "kafkaproxy.kafka.properties"; |
| public static final String KAFKA_PROXY_KAFKA_BROKERS = "kafkaproxy.kafka.brokers"; |
| |
| private static Map<String,String> DEFAULT_PROPERTIES = new HashMap<>(); |
| private static Map<String,String> CAN_OVERRIDE_DEFAULT_PROPERTIES = new HashMap<>(); |
| |
| |
| static { |
| DEFAULT_PROPERTIES.put("hbase.cluster.distributed","true"); |
| DEFAULT_PROPERTIES.put("zookeeper.znode.parent","/kafkaproxy"); |
| DEFAULT_PROPERTIES.put("hbase.regionserver.info.port","17010"); |
| DEFAULT_PROPERTIES.put("hbase.client.connection.impl", |
| "org.apache.hadoop.hbase.kafka.KafkaBridgeConnection"); |
| DEFAULT_PROPERTIES.put("hbase.regionserver.admin.service","false"); |
| DEFAULT_PROPERTIES.put("hbase.regionserver.client.service","false"); |
| DEFAULT_PROPERTIES.put("hbase.wal.provider", |
| "org.apache.hadoop.hbase.wal.DisabledWALProvider"); |
| DEFAULT_PROPERTIES.put("hbase.regionserver.workers","false"); |
| DEFAULT_PROPERTIES.put("hfile.block.cache.size","0.0001"); |
| DEFAULT_PROPERTIES.put("hbase.mob.file.cache.size","0"); |
| DEFAULT_PROPERTIES.put("hbase.masterless","true"); |
| DEFAULT_PROPERTIES.put("hbase.regionserver.metahandler.count","1"); |
| DEFAULT_PROPERTIES.put("hbase.regionserver.replication.handler.count","1"); |
| DEFAULT_PROPERTIES.put("hbase.regionserver.handler.count","1"); |
| DEFAULT_PROPERTIES.put("hbase.ipc.server.read.threadpool.size","3"); |
| |
| CAN_OVERRIDE_DEFAULT_PROPERTIES.put("hbase.regionserver.port","17020"); |
| } |
| |
| private static void printUsageAndExit(Options options, int exitCode) { |
| HelpFormatter formatter = new HelpFormatter(); |
| formatter.printHelp("hbase kafkaproxy start", "", options, |
| "\nTo run the kafka proxy as a daemon, execute " + |
| "hbase-connectors-daemon.sh start|stop kafkaproxy \n" + |
| "[--kafkabrokers (or -b) <kafka brokers (comma delmited)>] \n" + |
| "[--routerulesfile (or -r) <file with rules to route to kafka " |
| + "(defaults to kafka-route-rules.xm)>] \n" + |
| "[--kafkaproperties (or -f) <Path to properties file that " |
| + "has the kafka connection properties>] \n" + |
| "[--peername (or -p) name of hbase peer to use (defaults to hbasekafka)]\n " + |
| "[--znode (or -z) root znode (defaults to /kafkaproxy)] \n" + |
| "[--enablepeer (or -e) enable peer on startup (defaults to false)]\n " + |
| "[--auto (or -a) auto create peer] " + |
| "\n", true); |
| System.exit(exitCode); |
| } |
| |
| /** |
| * private constructor |
| */ |
| private KafkaProxy() { |
| |
| } |
| |
| /** |
| * Start the service |
| * @param args program arguments |
| * @throws Exception on error |
| */ |
| public static void main(String[] args) throws Exception { |
| |
| Map<String,String> otherProps = new HashMap<>(); |
| |
| Options options = new Options(); |
| |
| options.addRequiredOption("b", "kafkabrokers", true, |
| "Kafka Brokers (comma delimited)"); |
| options.addOption("r", "routerulesfile", true, |
| "file that has routing rules (defaults to conf/kafka-route-rules.xml"); |
| options.addOption("f", "kafkaproperties", true, |
| "Path to properties file that has the kafka connection properties"); |
| options.addRequiredOption("p", "peername", true, |
| "Name of hbase peer"); |
| options.addOption("z", "znode", true, |
| "root zode to use in zookeeper (defaults to /kafkaproxy)"); |
| options.addOption("a", "autopeer", false, |
| "Create a peer automatically to the hbase cluster"); |
| options.addOption("e", "enablepeer", false, |
| "enable peer on startup (defaults to false)"); |
| |
| LOG.info("STARTING executorService " + HRegionServer.class.getSimpleName()); |
| VersionInfo.logVersion(); |
| |
| Configuration conf = HBaseConfiguration.create(); |
| CommandLine commandLine = null; |
| |
| Configuration commandLineConf = new Configuration(); |
| commandLineConf.clear(); |
| |
| GenericOptionsParser parser = new GenericOptionsParser(commandLineConf, args); |
| String[] restArgs = parser.getRemainingArgs(); |
| |
| try { |
| commandLine = new BasicParser().parse(options, restArgs); |
| } catch (ParseException e) { |
| LOG.error("Could not parse: ", e); |
| printUsageAndExit(options, -1); |
| } |
| |
| |
| String peer=""; |
| if (!commandLine.hasOption('p')){ |
| System.err.println("hbase peer id is required"); |
| System.exit(-1); |
| } else { |
| peer = commandLine.getOptionValue('p'); |
| } |
| |
| boolean createPeer = false; |
| boolean enablePeer = false; |
| |
| if (commandLine.hasOption('a')){ |
| createPeer=true; |
| } |
| |
| if (commandLine.hasOption("a")){ |
| enablePeer=true; |
| } |
| |
| String rulesFile = StringUtils.defaultIfBlank( |
| commandLine.getOptionValue("r"),"kafka-route-rules.xml"); |
| |
| if (!new File(rulesFile).exists()){ |
| if (KafkaProxy.class.getClassLoader().getResource(rulesFile)!=null){ |
| rulesFile = KafkaProxy.class.getClassLoader().getResource(rulesFile).getFile(); |
| } else { |
| System.err.println("Rules file " + rulesFile + |
| " is invalid"); |
| System.exit(-1); |
| } |
| } |
| |
| otherProps.put(KafkaProxy.KAFKA_PROXY_RULES_FILE,rulesFile); |
| |
| if (commandLine.hasOption('f')){ |
| otherProps.put(KafkaProxy.KAFKA_PROXY_KAFKA_PROPERTIES,commandLine.getOptionValue('f')); |
| } else if (commandLine.hasOption('b')){ |
| otherProps.put(KafkaProxy.KAFKA_PROXY_KAFKA_BROKERS,commandLine.getOptionValue('b')); |
| } else { |
| System.err.println("Kafka connection properites or brokers must be specified"); |
| System.exit(-1); |
| } |
| |
| String zookeeperQ = conf.get("hbase.zookeeper.quorum") + ":" + |
| conf.get("hbase.zookeeper.property.clientPort"); |
| |
| ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(20000, 20); |
| |
| try (CuratorFramework zk = CuratorFrameworkFactory.newClient(zookeeperQ, retryPolicy); |
| ) { |
| zk.start(); |
| String rootZnode = "/kafkaproxy"; |
| setupZookeeperZnodes(zk,rootZnode,peer); |
| checkForOrCreateReplicationPeer(conf,zk,rootZnode,peer,createPeer,enablePeer); |
| } |
| |
| @SuppressWarnings("unchecked") |
| Class<? extends HRegionServer> regionServerClass = (Class<? extends HRegionServer>) conf |
| .getClass(HConstants.REGION_SERVER_IMPL, HRegionServer.class); |
| |
| List<String> allArgs = DEFAULT_PROPERTIES.keySet().stream() |
| .map((argKey)->("-D"+argKey+"="+ DEFAULT_PROPERTIES.get(argKey))) |
| .collect(Collectors.toList()); |
| |
| allArgs.addAll(CAN_OVERRIDE_DEFAULT_PROPERTIES.keySet().stream() |
| .filter((argKey)->commandLineConf.get(argKey,"").equalsIgnoreCase("")) |
| .map((argKey)->("-D"+argKey+"="+ CAN_OVERRIDE_DEFAULT_PROPERTIES.get(argKey))) |
| .collect(Collectors.toList())); |
| |
| for (Map.Entry<String,String> k : commandLineConf){ |
| allArgs.add("-D"+k.getKey()+"="+k.getValue()); |
| } |
| |
| otherProps.keySet().stream() |
| .map((argKey)->("-D"+argKey+"="+ otherProps.get(argKey))) |
| .forEach((item)->allArgs.add(item)); |
| |
| Arrays.stream(restArgs) |
| .filter((arg)->(arg.startsWith("-D")||arg.equals("start"))) |
| .forEach((arg)->allArgs.add(arg)); |
| |
| // is start there? |
| if (allArgs.stream() |
| .filter((arg)->arg.equalsIgnoreCase("start")).count() < 1){ |
| allArgs.add("start"); |
| } |
| |
| String[] newArgs=new String[allArgs.size()]; |
| allArgs.toArray(newArgs); |
| |
| new HRegionServerCommandLine(regionServerClass).doMain(newArgs); |
| } |
| |
| |
| /** |
| * Set up the needed znodes under the rootZnode |
| * @param zk CuratorFramework framework instance |
| * @param rootZnode Root znode |
| * @throws Exception If an error occurs |
| */ |
| public static void setupZookeeperZnodes(CuratorFramework zk, String rootZnode,String peer) |
| throws Exception { |
| // always gives the same uuid for the same name |
| UUID uuid = UUID.nameUUIDFromBytes(Bytes.toBytes(peer)); |
| String newValue = uuid.toString(); |
| byte []uuidBytes = Bytes.toBytes(newValue); |
| String idPath=rootZnode+"/hbaseid"; |
| if (zk.checkExists().forPath(idPath) == null) { |
| zk.create().forPath(rootZnode); |
| zk.create().forPath(rootZnode +"/hbaseid",uuidBytes); |
| } else { |
| // If the znode is there already make sure it has the |
| // expected value for the peer name. |
| byte[] znodeBytes = zk.getData().forPath(idPath).clone(); |
| if (!Bytes.equals(znodeBytes,uuidBytes)){ |
| String oldValue = Bytes.toString(znodeBytes); |
| LOG.warn("znode "+idPath+" has unexpected value "+ oldValue |
| +" expecting " + newValue + " " |
| + " (did the peer name for the proxy change?) " |
| + "Updating value"); |
| zk.setData().forPath(idPath, uuidBytes); |
| } |
| } |
| } |
| |
| /** |
| * Poll for the configured peer or create it if it does not exist |
| * (controlled by createIfMissing) |
| * @param hbaseConf the hbase configuratoin |
| * @param zk CuratorFramework object |
| * @param basePath base znode. |
| * @param peerName id if the peer to check for/create |
| * @param enablePeer if the peer is detected or created, enable it. |
| * @param createIfMissing if the peer doesn't exist, create it and peer to it. |
| */ |
| public static void checkForOrCreateReplicationPeer(Configuration hbaseConf, |
| CuratorFramework zk, |
| String basePath, |
| String peerName, boolean createIfMissing, |
| boolean enablePeer) { |
| try (Connection conn = ConnectionFactory.createConnection(hbaseConf); |
| Admin admin = conn.getAdmin()) { |
| |
| boolean peerThere = false; |
| |
| while (!peerThere) { |
| try { |
| ReplicationPeerConfig peerConfig = admin.getReplicationPeerConfig(peerName); |
| if (peerConfig !=null) { |
| peerThere=true; |
| } |
| } catch (ReplicationPeerNotFoundException e) { |
| if (createIfMissing) { |
| ReplicationPeerConfigBuilder builder = ReplicationPeerConfig.newBuilder(); |
| // get the current cluster's ZK config |
| String zookeeperQ = hbaseConf.get("hbase.zookeeper.quorum") + |
| ":" + |
| hbaseConf.get("hbase.zookeeper.property.clientPort"); |
| String znodePath = zookeeperQ + ":"+basePath; |
| ReplicationPeerConfig rconf = builder.setClusterKey(znodePath).build(); |
| admin.addReplicationPeer(peerName, rconf); |
| peerThere = true; |
| } |
| } |
| |
| if (peerThere) { |
| if (enablePeer){ |
| LOG.info("enable peer,{}", peerName); |
| List<ReplicationPeerDescription> peers = admin.listReplicationPeers().stream() |
| .filter(peer -> peer.getPeerId().equals(peerName)) |
| .filter(peer -> !peer.isEnabled()) |
| .collect(Collectors.toList()); |
| if (!peers.isEmpty()){ |
| admin.enableReplicationPeer(peerName); |
| } |
| } |
| break; |
| } else { |
| LOG.info("peer "+ |
| peerName+" not found, service will not completely start until the peer exists"); |
| } |
| Thread.sleep(5000); |
| } |
| |
| LOG.info("found replication peer " + peerName); |
| |
| } catch (Exception e) { |
| LOG.error("Exception running proxy ",e); |
| } |
| } |
| } |
