accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ktur...@apache.org
Subject [2/3] accumulo git commit: Merge branch '1.5' into 1.6
Date Fri, 13 Mar 2015 13:08:49 GMT
Merge branch '1.5' into 1.6

Conflicts:
	server/base/src/main/java/org/apache/accumulo/server/master/LiveTServerSet.java


Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/1ea90109
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/1ea90109
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/1ea90109

Branch: refs/heads/master
Commit: 1ea901091f6f443359cd46d5d71cb9eb9ca5db8f
Parents: 775cdb1 4ff11d4
Author: Keith Turner <kturner@apache.org>
Authored: Fri Mar 13 08:22:52 2015 -0400
Committer: Keith Turner <kturner@apache.org>
Committed: Fri Mar 13 08:22:52 2015 -0400

----------------------------------------------------------------------
 .../accumulo/server/master/LiveTServerSet.java  | 25 ++++++++++++++++----
 1 file changed, 20 insertions(+), 5 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/1ea90109/server/base/src/main/java/org/apache/accumulo/server/master/LiveTServerSet.java
----------------------------------------------------------------------
diff --cc server/base/src/main/java/org/apache/accumulo/server/master/LiveTServerSet.java
index 16f9885,0000000..332bfc9
mode 100644,000000..100644
--- a/server/base/src/main/java/org/apache/accumulo/server/master/LiveTServerSet.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/master/LiveTServerSet.java
@@@ -1,401 -1,0 +1,416 @@@
 +/*
 + * 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.accumulo.server.master;
 +
 +import static com.google.common.base.Charsets.UTF_8;
 +import static org.apache.accumulo.fate.zookeeper.ZooUtil.NodeMissingPolicy.SKIP;
 +
 +import java.nio.ByteBuffer;
 +import java.util.HashMap;
 +import java.util.HashSet;
 +import java.util.Map;
 +import java.util.Map.Entry;
 +import java.util.Set;
 +
 +import org.apache.accumulo.core.Constants;
 +import org.apache.accumulo.core.client.Instance;
 +import org.apache.accumulo.core.client.impl.thrift.ThriftSecurityException;
 +import org.apache.accumulo.core.conf.AccumuloConfiguration;
 +import org.apache.accumulo.core.data.KeyExtent;
 +import org.apache.accumulo.core.master.thrift.TabletServerStatus;
 +import org.apache.accumulo.core.tabletserver.thrift.NotServingTabletException;
 +import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
 +import org.apache.accumulo.core.util.AddressUtil;
 +import org.apache.accumulo.core.util.ServerServices;
 +import org.apache.accumulo.core.util.ThriftUtil;
 +import org.apache.accumulo.core.zookeeper.ZooUtil;
 +import org.apache.accumulo.server.master.state.TServerInstance;
 +import org.apache.accumulo.server.security.SystemCredentials;
 +import org.apache.accumulo.server.util.Halt;
 +import org.apache.accumulo.server.util.time.SimpleTimer;
 +import org.apache.accumulo.server.zookeeper.ZooCache;
 +import org.apache.accumulo.server.zookeeper.ZooLock;
 +import org.apache.accumulo.server.zookeeper.ZooReaderWriter;
 +import org.apache.accumulo.trace.instrument.Tracer;
 +import org.apache.hadoop.io.Text;
 +import org.apache.log4j.Logger;
 +import org.apache.thrift.TException;
 +import org.apache.thrift.transport.TTransport;
 +import org.apache.zookeeper.KeeperException;
 +import org.apache.zookeeper.KeeperException.NoNodeException;
 +import org.apache.zookeeper.KeeperException.NotEmptyException;
 +import org.apache.zookeeper.WatchedEvent;
 +import org.apache.zookeeper.Watcher;
 +import org.apache.zookeeper.data.Stat;
 +
 +import com.google.common.net.HostAndPort;
 +
 +public class LiveTServerSet implements Watcher {
 +
 +  public interface Listener {
 +    void update(LiveTServerSet current, Set<TServerInstance> deleted, Set<TServerInstance>
added);
 +  }
 +
 +  private static final Logger log = Logger.getLogger(LiveTServerSet.class);
 +
 +  private final Listener cback;
 +  private final Instance instance;
 +  private final AccumuloConfiguration conf;
 +  private ZooCache zooCache;
 +
 +  public class TServerConnection {
 +    private final HostAndPort address;
 +
 +    public TServerConnection(HostAndPort addr) throws TException {
 +      address = addr;
 +    }
 +
 +    private String lockString(ZooLock mlock) {
 +      return mlock.getLockID().serialize(ZooUtil.getRoot(instance) + Constants.ZMASTER_LOCK);
 +    }
 +
++    private void loadTablet(TabletClientService.Client client, ZooLock lock, KeyExtent extent)
throws TException {
++      client.loadTablet(Tracer.traceInfo(), SystemCredentials.get().toThrift(instance),
lockString(lock), extent.toThrift());
++    }
++
 +    public void assignTablet(ZooLock lock, KeyExtent extent) throws TException {
-       TabletClientService.Client client = ThriftUtil.getClient(new TabletClientService.Client.Factory(),
address, conf);
-       try {
-         client.loadTablet(Tracer.traceInfo(), SystemCredentials.get().toThrift(instance),
lockString(lock), extent.toThrift());
-       } finally {
-         ThriftUtil.returnClient(client);
++      if (extent.isMeta()) {
++        // see ACCUMULO-3597
++        TTransport transport = ThriftUtil.createTransport(address, conf);
++        try {
++          TabletClientService.Client client = ThriftUtil.createClient(new TabletClientService.Client.Factory(),
transport);
++          loadTablet(client, lock, extent);
++        } finally {
++          transport.close();
++        }
++      } else {
++        TabletClientService.Client client = ThriftUtil.getClient(new TabletClientService.Client.Factory(),
address, conf);
++        try {
++          loadTablet(client, lock, extent);
++        } finally {
++          ThriftUtil.returnClient(client);
++        }
 +      }
 +    }
 +
 +    public void unloadTablet(ZooLock lock, KeyExtent extent, boolean save) throws TException
{
 +      TabletClientService.Client client = ThriftUtil.getClient(new TabletClientService.Client.Factory(),
address, conf);
 +      try {
 +        client.unloadTablet(Tracer.traceInfo(), SystemCredentials.get().toThrift(instance),
lockString(lock), extent.toThrift(), save);
 +      } finally {
 +        ThriftUtil.returnClient(client);
 +      }
 +    }
 +
 +    public TabletServerStatus getTableMap(boolean usePooledConnection) throws TException,
ThriftSecurityException {
 +
 +      if (usePooledConnection == true)
 +        throw new UnsupportedOperationException();
 +
 +      TTransport transport = ThriftUtil.createTransport(address, conf);
 +
 +      try {
 +        TabletClientService.Client client = ThriftUtil.createClient(new TabletClientService.Client.Factory(),
transport);
 +        return client.getTabletServerStatus(Tracer.traceInfo(), SystemCredentials.get().toThrift(instance));
 +      } finally {
 +        if (transport != null)
 +          transport.close();
 +      }
 +    }
 +
 +    public void halt(ZooLock lock) throws TException, ThriftSecurityException {
 +      TabletClientService.Client client = ThriftUtil.getClient(new TabletClientService.Client.Factory(),
address, conf);
 +      try {
 +        client.halt(Tracer.traceInfo(), SystemCredentials.get().toThrift(instance), lockString(lock));
 +      } finally {
 +        ThriftUtil.returnClient(client);
 +      }
 +    }
 +
 +    public void fastHalt(ZooLock lock) throws TException {
 +      TabletClientService.Client client = ThriftUtil.getClient(new TabletClientService.Client.Factory(),
address, conf);
 +      try {
 +        client.fastHalt(Tracer.traceInfo(), SystemCredentials.get().toThrift(instance),
lockString(lock));
 +      } finally {
 +        ThriftUtil.returnClient(client);
 +      }
 +    }
 +
 +    public void flush(ZooLock lock, String tableId, byte[] startRow, byte[] endRow) throws
TException {
 +      TabletClientService.Client client = ThriftUtil.getClient(new TabletClientService.Client.Factory(),
address, conf);
 +      try {
 +        client.flush(Tracer.traceInfo(), SystemCredentials.get().toThrift(instance), lockString(lock),
tableId,
 +            startRow == null ? null : ByteBuffer.wrap(startRow), endRow == null ? null :
ByteBuffer.wrap(endRow));
 +      } finally {
 +        ThriftUtil.returnClient(client);
 +      }
 +    }
 +
 +    public void chop(ZooLock lock, KeyExtent extent) throws TException {
 +      TabletClientService.Client client = ThriftUtil.getClient(new TabletClientService.Client.Factory(),
address, conf);
 +      try {
 +        client.chop(Tracer.traceInfo(), SystemCredentials.get().toThrift(instance), lockString(lock),
extent.toThrift());
 +      } finally {
 +        ThriftUtil.returnClient(client);
 +      }
 +    }
 +
 +    public void splitTablet(ZooLock lock, KeyExtent extent, Text splitPoint) throws TException,
ThriftSecurityException, NotServingTabletException {
 +      TabletClientService.Client client = ThriftUtil.getClient(new TabletClientService.Client.Factory(),
address, conf);
 +      try {
 +        client.splitTablet(Tracer.traceInfo(), SystemCredentials.get().toThrift(instance),
extent.toThrift(),
 +            ByteBuffer.wrap(splitPoint.getBytes(), 0, splitPoint.getLength()));
 +      } finally {
 +        ThriftUtil.returnClient(client);
 +      }
 +    }
 +
 +    public void flushTablet(ZooLock lock, KeyExtent extent) throws TException {
 +      TabletClientService.Client client = ThriftUtil.getClient(new TabletClientService.Client.Factory(),
address, conf);
 +      try {
 +        client.flushTablet(Tracer.traceInfo(), SystemCredentials.get().toThrift(instance),
lockString(lock), extent.toThrift());
 +      } finally {
 +        ThriftUtil.returnClient(client);
 +      }
 +    }
 +
 +    public void compact(ZooLock lock, String tableId, byte[] startRow, byte[] endRow) throws
TException {
 +      TabletClientService.Client client = ThriftUtil.getClient(new TabletClientService.Client.Factory(),
address, conf);
 +      try {
 +        client.compact(Tracer.traceInfo(), SystemCredentials.get().toThrift(instance), lockString(lock),
tableId,
 +            startRow == null ? null : ByteBuffer.wrap(startRow), endRow == null ? null :
ByteBuffer.wrap(endRow));
 +      } finally {
 +        ThriftUtil.returnClient(client);
 +      }
 +    }
 +
 +    public boolean isActive(long tid) throws TException {
 +      TabletClientService.Client client = ThriftUtil.getClient(new TabletClientService.Client.Factory(),
address, conf);
 +      try {
 +        return client.isActive(Tracer.traceInfo(), tid);
 +      } finally {
 +        ThriftUtil.returnClient(client);
 +      }
 +    }
 +
 +  }
 +
 +  static class TServerInfo {
 +    TServerConnection connection;
 +    TServerInstance instance;
 +
 +    TServerInfo(TServerInstance instance, TServerConnection connection) {
 +      this.connection = connection;
 +      this.instance = instance;
 +    }
 +  };
 +
 +  // The set of active tservers with locks, indexed by their name in zookeeper
 +  private Map<String,TServerInfo> current = new HashMap<String,TServerInfo>();
 +  // as above, indexed by TServerInstance
 +  private Map<TServerInstance,TServerInfo> currentInstances = new HashMap<TServerInstance,TServerInfo>();
 +
 +  // The set of entries in zookeeper without locks, and the first time each was noticed
 +  private Map<String,Long> locklessServers = new HashMap<String,Long>();
 +
 +  public LiveTServerSet(Instance instance, AccumuloConfiguration conf, Listener cback) {
 +    this.cback = cback;
 +    this.instance = instance;
 +    this.conf = conf;
 +
 +  }
 +
 +  public synchronized ZooCache getZooCache() {
 +    if (zooCache == null)
 +      zooCache = new ZooCache(this);
 +    return zooCache;
 +  }
 +
 +  public synchronized void startListeningForTabletServerChanges() {
 +    scanServers();
 +    SimpleTimer.getInstance().schedule(new Runnable() {
 +      @Override
 +      public void run() {
 +        scanServers();
 +      }
 +    }, 0, 5000);
 +  }
 +
 +  public synchronized void scanServers() {
 +    try {
 +      final Set<TServerInstance> updates = new HashSet<TServerInstance>();
 +      final Set<TServerInstance> doomed = new HashSet<TServerInstance>();
 +
 +      final String path = ZooUtil.getRoot(instance) + Constants.ZTSERVERS;
 +
 +      HashSet<String> all = new HashSet<String>(current.keySet());
 +      all.addAll(getZooCache().getChildren(path));
 +
 +      locklessServers.keySet().retainAll(all);
 +
 +      for (String zPath : all) {
 +        checkServer(updates, doomed, path, zPath);
 +      }
 +
 +      // log.debug("Current: " + current.keySet());
 +      if (!doomed.isEmpty() || !updates.isEmpty())
 +        this.cback.update(this, doomed, updates);
 +    } catch (Exception ex) {
 +      log.error(ex, ex);
 +    }
 +  }
 +
 +  private void deleteServerNode(String serverNode) throws InterruptedException, KeeperException
{
 +    try {
 +      ZooReaderWriter.getInstance().delete(serverNode, -1);
 +    } catch (NotEmptyException ex) {
 +      // race condition: tserver created the lock after our last check; we'll see it at
the next check
 +    } catch (NoNodeException nne) {
 +      // someone else deleted it
 +    }
 +  }
 +
 +  private synchronized void checkServer(final Set<TServerInstance> updates, final
Set<TServerInstance> doomed, final String path, final String zPath)
 +      throws TException, InterruptedException, KeeperException {
 +
 +    TServerInfo info = current.get(zPath);
 +
 +    final String lockPath = path + "/" + zPath;
 +    Stat stat = new Stat();
 +    byte[] lockData = ZooLock.getLockData(getZooCache(), lockPath, stat);
 +
 +    if (lockData == null) {
 +      if (info != null) {
 +        doomed.add(info.instance);
 +        current.remove(zPath);
 +        currentInstances.remove(info.instance);
 +      }
 +
 +      Long firstSeen = locklessServers.get(zPath);
 +      if (firstSeen == null) {
 +        locklessServers.put(zPath, System.currentTimeMillis());
 +      } else if (System.currentTimeMillis() - firstSeen > 10 * 60 * 1000) {
 +        deleteServerNode(path + "/" + zPath);
 +        locklessServers.remove(zPath);
 +      }
 +    } else {
 +      locklessServers.remove(zPath);
 +      ServerServices services = new ServerServices(new String(lockData, UTF_8));
 +      HostAndPort client = services.getAddress(ServerServices.Service.TSERV_CLIENT);
 +      TServerInstance instance = new TServerInstance(client, stat.getEphemeralOwner());
 +
 +      if (info == null) {
 +        updates.add(instance);
 +        TServerInfo tServerInfo = new TServerInfo(instance, new TServerConnection(client));
 +        current.put(zPath, tServerInfo);
 +        currentInstances.put(instance, tServerInfo);
 +      } else if (!info.instance.equals(instance)) {
 +        doomed.add(info.instance);
 +        updates.add(instance);
 +        TServerInfo tServerInfo = new TServerInfo(instance, new TServerConnection(client));
 +        current.put(zPath, tServerInfo);
 +        currentInstances.remove(info.instance);
 +        currentInstances.put(instance, tServerInfo);
 +      }
 +    }
 +  }
 +
 +  @Override
 +  public void process(WatchedEvent event) {
 +
 +    // its important that these event are propagated by ZooCache, because this ensures when
reading zoocache that is has already processed the event and cleared
 +    // relevant nodes before code below reads from zoocache
 +
 +    if (event.getPath() != null) {
 +      if (event.getPath().endsWith(Constants.ZTSERVERS)) {
 +        scanServers();
 +      } else if (event.getPath().contains(Constants.ZTSERVERS)) {
 +        int pos = event.getPath().lastIndexOf('/');
 +
 +        // do only if ZTSERVER is parent
 +        if (pos >= 0 && event.getPath().substring(0, pos).endsWith(Constants.ZTSERVERS))
{
 +
 +          String server = event.getPath().substring(pos + 1);
 +
 +          final Set<TServerInstance> updates = new HashSet<TServerInstance>();
 +          final Set<TServerInstance> doomed = new HashSet<TServerInstance>();
 +
 +          final String path = ZooUtil.getRoot(instance) + Constants.ZTSERVERS;
 +
 +          try {
 +            checkServer(updates, doomed, path, server);
 +            if (!doomed.isEmpty() || !updates.isEmpty())
 +              this.cback.update(this, doomed, updates);
 +          } catch (Exception ex) {
 +            log.error(ex, ex);
 +          }
 +        }
 +      }
 +    }
 +  }
 +
 +  public synchronized TServerConnection getConnection(TServerInstance server) {
 +    if (server == null)
 +      return null;
 +    TServerInfo tServerInfo = currentInstances.get(server);
 +    if (tServerInfo == null)
 +      return null;
 +    return tServerInfo.connection;
 +  }
 +
 +  public synchronized Set<TServerInstance> getCurrentServers() {
 +    return new HashSet<TServerInstance>(currentInstances.keySet());
 +  }
 +
 +  public synchronized int size() {
 +    return current.size();
 +  }
 +
 +  public synchronized TServerInstance find(String tabletServer) {
 +    HostAndPort addr = AddressUtil.parseAddress(tabletServer, false);
 +    for (Entry<String,TServerInfo> entry : current.entrySet()) {
 +      if (entry.getValue().instance.getLocation().equals(addr))
 +        return entry.getValue().instance;
 +    }
 +    return null;
 +  }
 +
 +  public synchronized void remove(TServerInstance server) {
 +    String zPath = null;
 +    for (Entry<String,TServerInfo> entry : current.entrySet()) {
 +      if (entry.getValue().instance.equals(server)) {
 +        zPath = entry.getKey();
 +        break;
 +      }
 +    }
 +    if (zPath == null)
 +      return;
 +    current.remove(zPath);
 +    currentInstances.remove(server);
 +
 +    log.info("Removing zookeeper lock for " + server);
 +    String fullpath = ZooUtil.getRoot(instance) + Constants.ZTSERVERS + "/" + zPath;
 +    try {
 +      ZooReaderWriter.getInstance().recursiveDelete(fullpath, SKIP);
 +    } catch (Exception e) {
 +      String msg = "error removing tablet server lock";
 +      log.fatal(msg, e);
 +      Halt.halt(msg, -1);
 +    }
 +    getZooCache().clear(fullpath);
 +  }
 +}


Mime
View raw message