| /* |
| * 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.solr.cloud; |
| |
| import com.google.common.util.concurrent.AtomicLongMap; |
| |
| import org.apache.solr.SolrTestCaseJ4; |
| import org.apache.solr.common.cloud.SolrZkClient; |
| import org.apache.solr.common.cloud.ZkNodeProps; |
| import org.apache.solr.common.cloud.ZkStateReader; |
| import org.apache.solr.common.util.IOUtils; |
| import org.apache.solr.common.util.ObjectReleaseTracker; |
| import org.apache.solr.common.util.TimeSource; |
| import org.apache.solr.common.util.Utils; |
| import org.apache.solr.util.TimeOut; |
| import org.apache.zookeeper.CreateMode; |
| import org.apache.zookeeper.KeeperException; |
| import org.apache.zookeeper.Op; |
| import org.apache.zookeeper.WatchedEvent; |
| import org.apache.zookeeper.Watcher; |
| import org.apache.zookeeper.data.Stat; |
| import org.apache.zookeeper.jmx.ManagedUtil; |
| import org.apache.zookeeper.server.NIOServerCnxn; |
| import org.apache.zookeeper.server.NIOServerCnxnFactory; |
| import org.apache.zookeeper.server.ServerCnxn; |
| import org.apache.zookeeper.server.ServerCnxnFactory; |
| import org.apache.zookeeper.server.ServerConfig; |
| import org.apache.zookeeper.server.SessionTracker.Session; |
| import org.apache.zookeeper.server.ZKDatabase; |
| import org.apache.zookeeper.server.ZooKeeperServer; |
| import org.apache.zookeeper.server.persistence.FileTxnSnapLog; |
| import org.apache.zookeeper.server.quorum.QuorumPeerConfig.ConfigException; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import javax.management.JMException; |
| import java.io.BufferedReader; |
| import java.io.File; |
| import java.io.IOException; |
| import java.io.InputStreamReader; |
| import java.io.OutputStream; |
| import java.lang.invoke.MethodHandles; |
| import java.net.InetAddress; |
| import java.net.InetSocketAddress; |
| import java.net.Socket; |
| import java.net.UnknownHostException; |
| import java.nio.channels.SelectionKey; |
| import java.nio.channels.SocketChannel; |
| import java.nio.charset.StandardCharsets; |
| import java.nio.file.Path; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Comparator; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicReference; |
| |
| public class ZkTestServer { |
| |
| private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); |
| |
| public static File SOLRHOME; |
| static { |
| try { |
| SOLRHOME = new File(SolrTestCaseJ4.TEST_HOME()); |
| } catch (RuntimeException e) { |
| log.warn("TEST_HOME() does not exist - solrj test?"); |
| // solrj tests not working with TEST_HOME() |
| // must override getSolrHome |
| } |
| } |
| |
| public static final int TIMEOUT = 45000; |
| public static final int TICK_TIME = 1000; |
| |
| protected final ZKServerMain zkServer = new ZKServerMain(); |
| |
| private volatile Path zkDir; |
| |
| private volatile int clientPort; |
| |
| private volatile Thread zooThread; |
| |
| private volatile int theTickTime = TICK_TIME; |
| // SOLR-12101 - provide defaults to avoid max timeout 20 enforced by our server instance when tick time is 1000 |
| private volatile int maxSessionTimeout = 90000; |
| private volatile int minSessionTimeout = 3000; |
| |
| protected volatile SolrZkClient rootClient; |
| protected volatile SolrZkClient chRootClient; |
| |
| private volatile ZKDatabase zkDb; |
| |
| static public enum LimitViolationAction { |
| IGNORE, |
| REPORT, |
| FAIL, |
| } |
| |
| class ZKServerMain { |
| |
| private volatile ServerCnxnFactory cnxnFactory; |
| private volatile ZooKeeperServer zooKeeperServer; |
| private volatile LimitViolationAction violationReportAction = LimitViolationAction.REPORT; |
| private volatile WatchLimiter limiter = new WatchLimiter(1, LimitViolationAction.IGNORE); |
| |
| protected void initializeAndRun(String[] args) throws ConfigException, |
| IOException { |
| try { |
| ManagedUtil.registerLog4jMBeans(); |
| } catch (JMException e) { |
| log.warn("Unable to register log4j JMX control", e); |
| } |
| |
| ServerConfig config = new ServerConfig(); |
| if (args.length == 1) { |
| config.parse(args[0]); |
| } else { |
| config.parse(args); |
| } |
| |
| runFromConfig(config); |
| } |
| |
| private class WatchLimit { |
| private long limit; |
| private final String desc; |
| |
| private volatile LimitViolationAction action; |
| private AtomicLongMap<String> counters = AtomicLongMap.create(); |
| private ConcurrentHashMap<String,Long> maxCounters = new ConcurrentHashMap<>(); |
| |
| WatchLimit(long limit, String desc, LimitViolationAction action) { |
| this.limit = limit; |
| this.desc = desc; |
| this.action = action; |
| } |
| |
| public void setAction(LimitViolationAction action) { |
| this.action = action; |
| } |
| |
| public void setLimit(long limit) { |
| this.limit = limit; |
| } |
| |
| public void updateForWatch(String key, Watcher watcher) { |
| if (watcher != null) { |
| log.debug("Watch added: {}: {}", desc, key); |
| long count = counters.incrementAndGet(key); |
| Long lastCount = maxCounters.get(key); |
| if (lastCount == null || count > lastCount) { |
| maxCounters.put(key, count); |
| } |
| if (count > limit && action != LimitViolationAction.IGNORE) { |
| String msg = "Number of watches created in parallel for data: " + key + |
| ", type: " + desc + " exceeds limit (" + count + " > " + limit + ")"; |
| log.warn("{}", msg); |
| if (action == LimitViolationAction.FAIL) throw new AssertionError(msg); |
| } |
| } |
| } |
| |
| public void updateForFire(WatchedEvent event) { |
| if (log.isDebugEnabled()) { |
| log.debug("Watch fired: {}: {}", desc, event.getPath()); |
| } |
| counters.decrementAndGet(event.getPath()); |
| } |
| |
| private String reportLimitViolations() { |
| String[] maxKeys = maxCounters.keySet().toArray(new String[maxCounters.size()]); |
| Arrays.sort(maxKeys, new Comparator<String>() { |
| private final Comparator<Long> valComp = Comparator.<Long>naturalOrder().reversed(); |
| @Override |
| public int compare(String o1, String o2) { |
| return valComp.compare(maxCounters.get(o1), maxCounters.get(o2)); |
| } |
| }); |
| |
| StringBuilder sb = new StringBuilder(); |
| boolean first = true; |
| for (String key : maxKeys) { |
| long value = maxCounters.get(key); |
| if (value <= limit) continue; |
| if (first) { |
| sb.append("\nMaximum concurrent ").append(desc).append(" watches above limit:\n\n"); |
| first = false; |
| } |
| sb.append("\t").append(maxCounters.get(key)).append('\t').append(key).append('\n'); |
| } |
| return sb.toString(); |
| } |
| } |
| |
| public class WatchLimiter { |
| WatchLimit statLimit; |
| WatchLimit dataLimit; |
| WatchLimit childrenLimit; |
| |
| private WatchLimiter (long limit, LimitViolationAction action) { |
| statLimit = new WatchLimit(limit, "create/delete", action); |
| dataLimit = new WatchLimit(limit, "data", action); |
| childrenLimit = new WatchLimit(limit, "children", action); |
| } |
| |
| public void setAction(LimitViolationAction action) { |
| statLimit.setAction(action); |
| dataLimit.setAction(action); |
| childrenLimit.setAction(action); |
| } |
| |
| public void setLimit(long limit) { |
| statLimit.setLimit(limit); |
| dataLimit.setLimit(limit); |
| childrenLimit.setLimit(limit); |
| } |
| |
| public String reportLimitViolations() { |
| return statLimit.reportLimitViolations() + |
| dataLimit.reportLimitViolations() + |
| childrenLimit.reportLimitViolations(); |
| } |
| |
| private void updateForFire(WatchedEvent event) { |
| switch (event.getType()) { |
| case None: |
| break; |
| case NodeCreated: |
| case NodeDeleted: |
| statLimit.updateForFire(event); |
| break; |
| case NodeDataChanged: |
| dataLimit.updateForFire(event); |
| break; |
| case NodeChildrenChanged: |
| childrenLimit.updateForFire(event); |
| break; |
| case ChildWatchRemoved: |
| break; |
| case DataWatchRemoved: |
| break; |
| case PersistentWatchRemoved: |
| break; |
| } |
| } |
| } |
| |
| private class TestServerCnxn extends NIOServerCnxn { |
| |
| private final WatchLimiter limiter; |
| |
| public TestServerCnxn(ZooKeeperServer zk, SocketChannel sock, SelectionKey sk, |
| NIOServerCnxnFactory factory, WatchLimiter limiter) throws IOException { |
| super(zk, sock, sk, factory, null); |
| this.limiter = limiter; |
| } |
| |
| @Override |
| public synchronized void process(WatchedEvent event) { |
| limiter.updateForFire(event); |
| super.process(event); |
| } |
| } |
| |
| private class TestServerCnxnFactory extends NIOServerCnxnFactory { |
| |
| private final WatchLimiter limiter; |
| |
| public TestServerCnxnFactory(WatchLimiter limiter) throws IOException { |
| super(); |
| this.limiter = limiter; |
| } |
| } |
| |
| private class TestZKDatabase extends ZKDatabase { |
| |
| private final WatchLimiter limiter; |
| |
| public TestZKDatabase(FileTxnSnapLog snapLog, WatchLimiter limiter) { |
| super(snapLog); |
| this.limiter = limiter; |
| } |
| |
| @Override |
| public Stat statNode(String path, ServerCnxn serverCnxn) throws KeeperException.NoNodeException { |
| limiter.statLimit.updateForWatch(path, serverCnxn); |
| return super.statNode(path, serverCnxn); |
| } |
| |
| @Override |
| public byte[] getData(String path, Stat stat, Watcher watcher) throws KeeperException.NoNodeException { |
| limiter.dataLimit.updateForWatch(path, watcher); |
| return super.getData(path, stat, watcher); |
| } |
| |
| @Override |
| public List<String> getChildren(String path, Stat stat, Watcher watcher) throws KeeperException.NoNodeException { |
| limiter.childrenLimit.updateForWatch(path, watcher); |
| return super.getChildren(path, stat, watcher); |
| } |
| } |
| |
| /** |
| * Run from a ServerConfig. |
| * @param config ServerConfig to use. |
| * @throws IOException If there is a low-level I/O error. |
| */ |
| public void runFromConfig(ServerConfig config) throws IOException { |
| ObjectReleaseTracker.track(this); |
| log.info("Starting server"); |
| try { |
| // ZooKeeper maintains a static collection of AuthenticationProviders, so |
| // we make sure the SASL provider is loaded so that it can be used in |
| // subsequent tests. |
| System.setProperty("zookeeper.authProvider.1", |
| "org.apache.zookeeper.server.auth.SASLAuthenticationProvider"); |
| // Note that this thread isn't going to be doing anything else, |
| // so rather than spawning another thread, we will just call |
| // run() in this thread. |
| // create a file logger url from the command line args |
| FileTxnSnapLog ftxn = new FileTxnSnapLog(config.getDataLogDir(), config.getDataDir()); |
| |
| zooKeeperServer = new ZooKeeperServer(ftxn, config.getTickTime(), |
| config.getMinSessionTimeout(), config.getMaxSessionTimeout(), |
| config.getClientPortListenBacklog(), |
| new TestZKDatabase(ftxn, limiter), ""); |
| cnxnFactory = new TestServerCnxnFactory(limiter); |
| cnxnFactory.configure(config.getClientPortAddress(), |
| config.getMaxClientCnxns()); |
| cnxnFactory.startup(zooKeeperServer); |
| cnxnFactory.join(); |
| |
| if (violationReportAction != LimitViolationAction.IGNORE) { |
| String limitViolations = limiter.reportLimitViolations(); |
| if (!limitViolations.isEmpty()) { |
| log.warn("Watch limit violations: {}", limitViolations); |
| if (violationReportAction == LimitViolationAction.FAIL) { |
| throw new AssertionError("Parallel watch limits violated"); |
| } |
| } |
| } |
| } catch (InterruptedException e) { |
| // warn, but generally this is ok |
| log.warn("Server interrupted", e); |
| } |
| } |
| |
| /** |
| * Shutdown the serving instance |
| * @throws IOException If there is a low-level I/O error. |
| */ |
| protected void shutdown() throws IOException { |
| |
| // shutting down the cnxnFactory will close the zooKeeperServer |
| // zooKeeperServer.shutdown(); |
| |
| ZKDatabase zkDb = zooKeeperServer.getZKDatabase(); |
| try { |
| if (cnxnFactory != null) { |
| while (true) { |
| cnxnFactory.shutdown(); |
| try { |
| cnxnFactory.join(); |
| break; |
| } catch (InterruptedException e) { |
| // Thread.currentThread().interrupt(); |
| // don't keep interrupted status |
| } |
| } |
| } |
| if (zkDb != null) { |
| zkDb.close(); |
| } |
| |
| if (cnxnFactory != null) { |
| try { |
| int port = cnxnFactory.getLocalPort(); |
| if (port > 0) { |
| waitForServerDown(getZkHost(), 30000); |
| } |
| } catch (NullPointerException ignored) { |
| // server never successfully started |
| } |
| } |
| } finally { |
| ObjectReleaseTracker.release(this); |
| } |
| } |
| |
| public int getLocalPort() { |
| if (cnxnFactory == null) { |
| throw new IllegalStateException("A port has not yet been selected"); |
| } |
| int port; |
| try { |
| port = cnxnFactory.getLocalPort(); |
| } catch (NullPointerException e) { |
| throw new IllegalStateException("A port has not yet been selected"); |
| } |
| if (port == 0) { |
| throw new IllegalStateException("A port has not yet been selected"); |
| } |
| return port; |
| } |
| |
| public void setViolationReportAction(LimitViolationAction violationReportAction) { |
| this.violationReportAction = violationReportAction; |
| } |
| |
| public WatchLimiter getLimiter() { |
| return limiter; |
| } |
| } |
| |
| public ZkTestServer(Path zkDir) throws Exception { |
| this(zkDir, 0); |
| } |
| |
| public ZkTestServer(Path zkDir, int port) throws KeeperException, InterruptedException { |
| this.zkDir = zkDir; |
| this.clientPort = port; |
| String reportAction = System.getProperty("tests.zk.violationReportAction"); |
| if (reportAction != null) { |
| log.info("Overriding violation report action to: {}", reportAction); |
| setViolationReportAction(LimitViolationAction.valueOf(reportAction)); |
| } |
| String limiterAction = System.getProperty("tests.zk.limiterAction"); |
| if (limiterAction != null) { |
| log.info("Overriding limiter action to: {}", limiterAction); |
| getLimiter().setAction(LimitViolationAction.valueOf(limiterAction)); |
| } |
| |
| ObjectReleaseTracker.track(this); |
| } |
| |
| private void init(boolean solrFormat) throws Exception { |
| try { |
| rootClient = new SolrZkClient(getZkHost(), TIMEOUT, 30000); |
| } catch (Exception e) { |
| log.error("error making rootClient, trying one more time", e); |
| rootClient = new SolrZkClient(getZkHost(), TIMEOUT, 30000); |
| } |
| |
| if (solrFormat) { |
| tryCleanSolrZkNode(); |
| makeSolrZkNode(); |
| } |
| |
| chRootClient = new SolrZkClient(getZkAddress(), AbstractZkTestCase.TIMEOUT, 30000); |
| } |
| |
| public String getZkHost() { |
| String hostName = System.getProperty("hostName", "127.0.0.1"); |
| return hostName + ":" + zkServer.getLocalPort(); |
| } |
| |
| public String getZkAddress() { |
| return getZkAddress("/solr"); |
| } |
| |
| /** |
| * Get a connection string for this server with a given chroot |
| * @param chroot the chroot |
| * @return the connection string |
| */ |
| public String getZkAddress(String chroot) { |
| if (!chroot.startsWith("/")) { |
| chroot = "/" + chroot; |
| } |
| return getZkHost() + chroot; |
| } |
| |
| /** |
| * Check that a path exists in this server |
| * @param path the path to check |
| * @throws IOException if an IO exception occurs |
| */ |
| public void ensurePathExists(String path) throws IOException { |
| try (SolrZkClient client = new SolrZkClient(getZkHost(), 10000)) { |
| client.makePath(path, null, CreateMode.PERSISTENT, null, false, true, 0); |
| } catch (InterruptedException | KeeperException e) { |
| e.printStackTrace(); |
| throw new IOException("Error checking path " + path, SolrZkClient.checkInterrupted(e)); |
| } |
| } |
| |
| public int getPort() { |
| return zkServer.getLocalPort(); |
| } |
| |
| public void expire(final long sessionId) { |
| zkServer.zooKeeperServer.expire(new Session() { |
| @Override |
| public long getSessionId() { |
| return sessionId; |
| } |
| |
| @Override |
| public int getTimeout() { |
| return 4000; |
| } |
| |
| @Override |
| public boolean isClosing() { |
| return false; |
| } |
| }); |
| } |
| |
| public ZKDatabase getZKDatabase() { |
| return zkServer.zooKeeperServer.getZKDatabase(); |
| } |
| |
| public void setZKDatabase(ZKDatabase zkDb) { |
| this.zkDb = zkDb; |
| zkServer.zooKeeperServer.setZKDatabase(zkDb); |
| } |
| |
| public void run() throws InterruptedException, IOException { |
| run(true); |
| } |
| |
| public void run(boolean solrFormat) throws InterruptedException, IOException { |
| log.info("STARTING ZK TEST SERVER"); |
| AtomicReference<Throwable> zooError = new AtomicReference<>(); |
| try { |
| if (zooThread != null) { |
| throw new IllegalStateException("ZK TEST SERVER IS ALREADY RUNNING"); |
| } |
| Thread parentThread = Thread.currentThread(); |
| // we don't call super.distribSetUp |
| zooThread = new Thread("ZkTestServer Run Thread") { |
| |
| @Override |
| public void run() { |
| ServerConfig config = new ServerConfig() { |
| |
| { |
| setClientPort(ZkTestServer.this.clientPort); |
| this.dataDir = zkDir.toFile(); |
| this.dataLogDir = zkDir.toFile(); |
| this.tickTime = theTickTime; |
| this.maxSessionTimeout = ZkTestServer.this.maxSessionTimeout; |
| this.minSessionTimeout = ZkTestServer.this.minSessionTimeout; |
| } |
| |
| public void setClientPort(int clientPort) { |
| if (clientPortAddress != null) { |
| try { |
| this.clientPortAddress = new InetSocketAddress( |
| InetAddress.getByName(clientPortAddress.getHostName()), clientPort); |
| } catch (UnknownHostException e) { |
| throw new RuntimeException(e); |
| } |
| } else { |
| this.clientPortAddress = new InetSocketAddress(clientPort); |
| } |
| log.info("client port: {}", this.clientPortAddress); |
| } |
| }; |
| try { |
| zkServer.runFromConfig(config); |
| } catch (Throwable t) { |
| zooError.set(t); |
| parentThread.interrupt(); |
| } |
| } |
| }; |
| |
| ObjectReleaseTracker.track(zooThread); |
| zooThread.start(); |
| |
| int cnt = 0; |
| int port = -1; |
| try { |
| port = getPort(); |
| } catch (IllegalStateException ignored) { |
| // Possibly fix this API to return null instead of throwing |
| } |
| while (port < 1) { |
| Thread.sleep(100); |
| try { |
| port = getPort(); |
| } catch (IllegalStateException ignored) { |
| // Possibly fix this API to return null instead of throwing |
| } |
| if (cnt == 500) { |
| throw new RuntimeException("Could not get the port for ZooKeeper server"); |
| } |
| cnt++; |
| } |
| log.info("start zk server on port: {}", port); |
| |
| waitForServerUp(getZkHost(), 30000); |
| |
| init(solrFormat); |
| } catch (Exception e) { |
| RuntimeException toThrow = new RuntimeException("Could not get ZK port"); |
| Throwable t = zooError.get(); |
| if (t != null) { |
| toThrow.initCause(t); |
| toThrow.addSuppressed(e); |
| } else { |
| toThrow.initCause(e); |
| } |
| throw toThrow; |
| } |
| } |
| |
| public void shutdown() throws IOException, InterruptedException { |
| log.info("Shutting down ZkTestServer."); |
| try { |
| IOUtils.closeQuietly(rootClient); |
| IOUtils.closeQuietly(chRootClient); |
| } finally { |
| |
| // TODO: this can log an exception while trying to unregister a JMX MBean |
| try { |
| zkServer.shutdown(); |
| } catch (Exception e) { |
| log.error("Exception shutting down ZooKeeper Test Server",e); |
| } |
| |
| if (zkDb != null) { |
| zkDb.close(); |
| } |
| |
| while (true) { |
| try { |
| zooThread.join(); |
| ObjectReleaseTracker.release(zooThread); |
| zooThread = null; |
| break; |
| } catch (InterruptedException e) { |
| // don't keep interrupted status |
| } catch (NullPointerException e) { |
| // okay |
| break; |
| } |
| } |
| } |
| ObjectReleaseTracker.release(this); |
| } |
| |
| public static boolean waitForServerDown(String hp, long timeoutMs) { |
| log.info("waitForServerDown: {}", hp); |
| final TimeOut timeout = new TimeOut(timeoutMs, TimeUnit.MILLISECONDS, TimeSource.NANO_TIME); |
| while (true) { |
| try { |
| HostPort hpobj = parseHostPortList(hp).get(0); |
| send4LetterWord(hpobj.host, hpobj.port, "stat"); |
| } catch (IOException e) { |
| return true; |
| } |
| |
| if (timeout.hasTimedOut()) { |
| throw new RuntimeException("Time out waiting for ZooKeeper shutdown!"); |
| } |
| try { |
| Thread.sleep(250); |
| } catch (InterruptedException e) { |
| // ignore |
| } |
| } |
| } |
| |
| public static boolean waitForServerUp(String hp, long timeoutMs) { |
| log.info("waitForServerUp: {}", hp); |
| final TimeOut timeout = new TimeOut(timeoutMs, TimeUnit.MILLISECONDS, TimeSource.NANO_TIME); |
| while (true) { |
| try { |
| HostPort hpobj = parseHostPortList(hp).get(0); |
| send4LetterWord(hpobj.host, hpobj.port, "stat"); |
| return true; |
| } catch (IOException e) { |
| e.printStackTrace(); |
| } |
| |
| if (timeout.hasTimedOut()) { |
| throw new RuntimeException("Time out waiting for ZooKeeper to startup!"); |
| } |
| try { |
| Thread.sleep(250); |
| } catch (InterruptedException e) { |
| // ignore |
| } |
| } |
| } |
| |
| public static class HostPort { |
| String host; |
| int port; |
| |
| HostPort(String host, int port) { |
| assert !host.contains(":") : host; |
| this.host = host; |
| this.port = port; |
| } |
| } |
| |
| /** |
| * Send the 4letterword |
| * @param host the destination host |
| * @param port the destination port |
| * @param cmd the 4letterword |
| * @return server response |
| |
| */ |
| public static String send4LetterWord(String host, int port, String cmd) |
| throws IOException |
| { |
| log.info("connecting to {} {}", host, port); |
| BufferedReader reader = null; |
| try (Socket sock = new Socket(host, port)) { |
| OutputStream outstream = sock.getOutputStream(); |
| outstream.write(cmd.getBytes(StandardCharsets.US_ASCII)); |
| outstream.flush(); |
| // this replicates NC - close the output stream before reading |
| sock.shutdownOutput(); |
| |
| reader = new BufferedReader( |
| new InputStreamReader(sock.getInputStream(), StandardCharsets.US_ASCII)); |
| StringBuilder sb = new StringBuilder(); |
| String line; |
| while ((line = reader.readLine()) != null) { |
| sb.append(line).append("\n"); |
| } |
| return sb.toString(); |
| } finally { |
| if (reader != null) { |
| reader.close(); |
| } |
| } |
| } |
| |
| public static List<HostPort> parseHostPortList(String hplist) { |
| log.info("parse host and port list: {}", hplist); |
| ArrayList<HostPort> alist = new ArrayList<>(); |
| for (String hp : hplist.split(",")) { |
| int idx = hp.lastIndexOf(':'); |
| String host = hp.substring(0, idx); |
| int port; |
| try { |
| port = Integer.parseInt(hp.substring(idx + 1)); |
| } catch (RuntimeException e) { |
| throw new RuntimeException("Problem parsing " + hp + e.toString()); |
| } |
| alist.add(new HostPort(host, port)); |
| } |
| return alist; |
| } |
| |
| public int getTheTickTime() { |
| return theTickTime; |
| } |
| |
| public void setTheTickTime(int theTickTime) { |
| this.theTickTime = theTickTime; |
| } |
| |
| public Path getZkDir() { |
| return zkDir; |
| } |
| |
| public void setViolationReportAction(LimitViolationAction violationReportAction) { |
| zkServer.setViolationReportAction(violationReportAction); |
| } |
| |
| public ZKServerMain.WatchLimiter getLimiter() { |
| return zkServer.getLimiter(); |
| } |
| |
| public int getMaxSessionTimeout() { |
| return maxSessionTimeout; |
| } |
| |
| public int getMinSessionTimeout() { |
| return minSessionTimeout; |
| } |
| |
| public void setMaxSessionTimeout(int maxSessionTimeout) { |
| this.maxSessionTimeout = maxSessionTimeout; |
| } |
| |
| public void setMinSessionTimeout(int minSessionTimeout) { |
| this.minSessionTimeout = minSessionTimeout; |
| } |
| |
| void buildZooKeeper(String config, |
| String schema) throws Exception { |
| buildZooKeeper(SOLRHOME, config, schema); |
| } |
| |
| public static void putConfig(String confName, SolrZkClient zkClient, File solrhome, final String name) |
| throws Exception { |
| putConfig(confName, zkClient, null, solrhome, name, name); |
| } |
| |
| |
| public static void putConfig(String confName, SolrZkClient zkClient, File solrhome, final String srcName, |
| String destName) throws Exception { |
| putConfig(confName, zkClient, null, solrhome, srcName, destName); |
| } |
| |
| public static void putConfig(String confName, SolrZkClient zkClient, String zkChroot, File solrhome, |
| final String srcName, String destName) throws Exception { |
| File file = new File(solrhome, "collection1" |
| + File.separator + "conf" + File.separator + srcName); |
| if (!file.exists()) { |
| if (log.isInfoEnabled()) { |
| log.info("skipping {} because it doesn't exist", file.getAbsolutePath()); |
| } |
| return; |
| } |
| |
| String destPath = "/configs/" + confName + "/" + destName; |
| if (zkChroot != null) { |
| destPath = zkChroot + destPath; |
| } |
| if (log.isInfoEnabled()) { |
| log.info("put {} to {}", file.getAbsolutePath(), destPath); |
| } |
| zkClient.makePath(destPath, file, false, true); |
| } |
| |
| // static to share with distrib test |
| public void buildZooKeeper(File solrhome, String config, String schema) throws Exception { |
| |
| Map<String,Object> props = new HashMap<>(); |
| props.put("configName", "conf1"); |
| final ZkNodeProps zkProps = new ZkNodeProps(props); |
| |
| |
| List<Op> ops = new ArrayList<>(2); |
| String path = "/collections"; |
| ops.add(Op.create(path, null, chRootClient.getZkACLProvider().getACLsToAdd(path), CreateMode.PERSISTENT)); |
| path = "/collections/collection1"; |
| ops.add(Op.create(path, Utils.toJSON(zkProps), chRootClient.getZkACLProvider().getACLsToAdd(path), CreateMode.PERSISTENT)); |
| path = "/collections/collection1/shards"; |
| ops.add(Op.create(path, null, chRootClient.getZkACLProvider().getACLsToAdd(path), CreateMode.PERSISTENT)); |
| path = "/collections/control_collection"; |
| ops.add(Op.create(path, Utils.toJSON(zkProps), chRootClient.getZkACLProvider().getACLsToAdd(path), CreateMode.PERSISTENT)); |
| path = "/collections/control_collection/shards"; |
| ops.add(Op.create(path, null, chRootClient.getZkACLProvider().getACLsToAdd(path), CreateMode.PERSISTENT)); |
| path = "/configs"; |
| ops.add(Op.create(path, null, chRootClient.getZkACLProvider().getACLsToAdd(path), CreateMode.PERSISTENT)); |
| path = "/configs/conf1"; |
| ops.add(Op.create(path, null, chRootClient.getZkACLProvider().getACLsToAdd(path), CreateMode.PERSISTENT)); |
| chRootClient.multi(ops, true); |
| |
| // this workaround is acceptable until we remove legacyCloud because we just init a single core here |
| String defaultClusterProps = "{\""+ZkStateReader.LEGACY_CLOUD+"\":\"true\"}"; |
| chRootClient.makePath(ZkStateReader.CLUSTER_PROPS, defaultClusterProps.getBytes(StandardCharsets.UTF_8), CreateMode.PERSISTENT, true); |
| // for now, always upload the config and schema to the canonical names |
| putConfig("conf1", chRootClient, solrhome, config, "solrconfig.xml"); |
| putConfig("conf1", chRootClient, solrhome, schema, "schema.xml"); |
| |
| putConfig("conf1", chRootClient, solrhome, "solrconfig.snippet.randomindexconfig.xml"); |
| putConfig("conf1", chRootClient, solrhome, "stopwords.txt"); |
| putConfig("conf1", chRootClient, solrhome, "protwords.txt"); |
| putConfig("conf1", chRootClient, solrhome, "currency.xml"); |
| putConfig("conf1", chRootClient, solrhome, "enumsConfig.xml"); |
| putConfig("conf1", chRootClient, solrhome, "open-exchange-rates.json"); |
| putConfig("conf1", chRootClient, solrhome, "mapping-ISOLatin1Accent.txt"); |
| putConfig("conf1", chRootClient, solrhome, "old_synonyms.txt"); |
| putConfig("conf1", chRootClient, solrhome, "synonyms.txt"); |
| } |
| |
| public void makeSolrZkNode() throws Exception { |
| rootClient.makePath("/solr", false, true); |
| } |
| |
| public void tryCleanSolrZkNode() throws Exception { |
| tryCleanPath("/solr"); |
| } |
| |
| void tryCleanPath(String path) throws Exception { |
| if (rootClient.exists(path, true)) { |
| rootClient.clean(path); |
| } |
| } |
| |
| protected void printLayout() throws Exception { |
| rootClient.printLayoutToStdOut(); |
| } |
| |
| public SolrZkClient getZkClient() { |
| return chRootClient; |
| } |
| } |
