Merge lp:~nchohan/appscale/appscale-1.6 into lp:appscale

[Navraj Chohan]

Tested on 1 node for Cassandra, HBase, Hypertable, Mysql, RedisDB, mongodb
Tested on 3 nodes for the same.
Tested with multiple applications with Java/Go/Python.

New items:
RabbitMQ for the task queue system.
HBase 0.90.4
Cassandra 1.0.7
Hypertable 0.9.5.5

GAE 1.6.1 for python and go

Some bug fixes

[Navraj Chohan (nchohan)]

Updated hadoop to 0.20.2-cdh3u3. HBase is also cdh3u3.

Fixed a security issue with task queue request.

[Chris Bunch (cgb-cs)]

Testing now - works on 2 node Cassandra simple deployment with pipeline-test sample app. Continuing with other apps / deployments.

[Chris Bunch (cgb-cs)]

A few style comments:

In rabbitmq.rb:

1) On line 11, remove the trailing space.

2) On line 15, shouldn't the semi-colons be double ampersands? Don't you want to only exec the second command if the first succeeds?

3) Lines 20-21 are commented out since you're not using god to manage rabbitmq - thus, remove those and 16-18, since you're defining variables that you're not using.

4) On 11, 25, and 54, you don't need to pass in the secret - functions can get it with HelperFunctions.get_secret()

5) On line 30, extract the rabbitmq port to a global constant

6) Remove line 41, since you don't use it.

7) Remove line 50

8) Remove line 63

review: Needs Fixing

[Chris Bunch (cgb-cs)]

More style stuff:

djinn_job_data, nginx, and pbserver looks good to me

In haproxy.rb, you've got these lines:

+ # Timeout a request if Mongrel does not accept a connection for 60 seconds
+ timeout connect 60000

which seem to indicate that the value is msec - 60000 msec -> 60 sec

but this doesn't seem to hold for the other vals:

+ # Timeout a request if the client did not read any data for 240 seconds
+ timeout client 60000

why isn't the above 60 seconds as well?

In terminate.rb, this whole block:

+#begin
+# require 'rabbitmq'
+# RabbitMQ.stop
+#rescue Execption
+# puts "Problem with rabbitmq, moving on"
+#end
+

is commented out, so just delete it.

In djinn.rb:

1) Remove "Djinn.rb" from all log_debug statements, since it isn't consistently applied (and other files don't identify themselves) and isn't standard for us.

review: Needs Fixing

[Navraj Chohan (nchohan)]

Fixes committed.

review: Needs Resubmitting

[Chris Bunch (cgb-cs)]

review: Approve

[lp:~nchohan/appscale/appscale-1.6]

790. By Navraj Chohan on 2012-02-17

RabbitMQ for the task queue system. HBase 0.90.4, Cassandra 1.0.7, Hypertable 0.9.5.5, GAE 1.6.1 for python and go, and other bug fixes

=== modified file 'AppController/djinn.rb'
--- AppController/djinn.rb	2012-02-01 19:17:26 +0000
+++ AppController/djinn.rb	2012-02-16 06:02:21 +0000
@@ -22,6 +22,7 @@
 require 'nginx'
 require 'pbserver'
 require 'blobstore'
+require 'rabbitmq'
 require 'app_controller_client'
 require 'user_app_client'
 require 'ejabberd'
@@ -188,7 +189,7 @@
       return "Error: Credential format wrong"
     end
 
-    Djinn.log_debug("Parameters were valid")
+    Djinn.log_debug("Djinn (set_parameters) Parameters were valid")
 
     keyname = possible_credentials["keyname"]
     @nodes = Djinn.convert_location_array_to_class(djinn_locations, keyname)
@@ -198,15 +199,15 @@
     convert_fqdns_to_ips
     @creds = sanitize_credentials
 
-    Djinn.log_debug("Djinn locations: #{@nodes.join(', ')}")
-    Djinn.log_debug("DB Credentials: #{HelperFunctions.obscure_creds(@creds).inspect}")
+    Djinn.log_debug("Djinn (set_parameters) locations: #{@nodes.join(', ')}")
+    Djinn.log_debug("Djinn (set_parameters) DB Credentials: #{HelperFunctions.obscure_creds(@creds).inspect}")
     Djinn.log_debug("Apps to load: #{@app_names.join(', ')}")
 
     find_me_in_locations
     if @my_index == nil
       return "Error: Couldn't find me in the node map"
     end
-    Djinn.log_debug("My index = #{@my_index}")
+    Djinn.log_debug("Djinn (set_parameters) My index = #{@my_index}")
 
     ENV['EC2_URL'] = @creds['ec2_url']
     
@@ -223,7 +224,7 @@
     end
 
     @app_names = app_names
-    return "app names is now #{@app_names.join(', ')}"
+    return "App names is now #{@app_names.join(', ')}"
   end
  
   def status(secret)
@@ -284,8 +285,9 @@
   def stop_app(app_name, secret)
     return BAD_SECRET_MSG unless valid_secret?(secret)
     app_name.gsub!(/[^\w\d\-]/, "")
-    Djinn.log_debug("Shutting down app named [#{app_name}]")
+    Djinn.log_debug("Djinn (stop_app): Shutting down app named [#{app_name}]")
     result = ""
+    Djinn.log_run("rm -rf /var/apps/#{app_name}")
    
     # app shutdown process can take more than 30 seconds
     # so run it in a new thread to avoid 'execution expired'
@@ -299,7 +301,7 @@
             ip = node.private_ip
             acc = AppControllerClient.new(ip, @@secret)
             result = acc.stop_app(app_name)
-            Djinn.log_debug("Removing application #{app_name} --- #{ip} returned #{result} (#{result.class})")
+            Djinn.log_debug("Djinn (stop_app): Removing application #{app_name} --- #{ip} returned #{result} (#{result.class})")
           end
         }
       end
@@ -309,7 +311,7 @@
         ip = HelperFunctions.read_file("#{APPSCALE_HOME}/.appscale/masters")
         uac = UserAppClient.new(ip, @@secret)
         result = uac.delete_app(app_name)
-        Djinn.log_debug("Delete app: #{ip} returned #{result} (#{result.class})")
+        Djinn.log_debug("Djinn (stop_app) Delete app: #{ip} returned #{result} (#{result.class})")
       end
      
       # may need to stop XMPP listener
@@ -364,7 +366,7 @@
       ip = @nodes[index].private_ip
       acc = AppControllerClient.new(ip, @@secret)
       result = acc.set_apps(apps)
-      Djinn.log_debug("#{ip} returned #{result} (#{result.class})")
+      Djinn.log_debug("djinn.rb:update #{ip} returned #{result} (#{result.class})")
       @everyone_else_is_done = false if !result
     }
 
@@ -706,16 +708,16 @@
     infrastructure = @creds["infrastructure"]    
  
     if is_hybrid_cloud?
-      Djinn.log_debug("Getting hybrid ips with creds #{@creds.inspect}")
+      Djinn.log_debug("Djinn (set_uaserver_ips): Getting hybrid ips with creds #{@creds.inspect}")
       public_ips, private_ips = HelperFunctions.get_hybrid_ips(@creds)
     else
-      Djinn.log_debug("Getting cloud ips for #{infrastructure} with keyname #{keyname}")
+      Djinn.log_debug("Djinn (set_uaserver_ips): Getting cloud ips for #{infrastructure} with keyname #{keyname}")
       public_ips, private_ips = HelperFunctions.get_cloud_ips(infrastructure, keyname)
     end
  
-    Djinn.log_debug("Public ips are #{public_ips.join(', ')}")
-    Djinn.log_debug("Private ips are #{private_ips.join(', ')}")
-    Djinn.log_debug("Looking for #{ip_addr}")
+    Djinn.log_debug("Djinn (set_uaserver_ips): Public ips are #{public_ips.join(', ')}")
+    Djinn.log_debug("Djinn (set_uaserver_ips): Private ips are #{private_ips.join(', ')}")
+    Djinn.log_debug("Djinn (set_uaserver_ips): Looking for #{ip_addr}")
 
     public_ips.each_index { |index|
       node_public_ip = HelperFunctions.convert_fqdn_to_ip(public_ips[index])
@@ -747,16 +749,16 @@
     infrastructure = @creds["infrastructure"]    
 
     if is_hybrid_cloud?
-      Djinn.log_debug("Getting hybrid ips with creds #{@creds.inspect}")
+      Djinn.log_debug("Djinn (set_uaserver_ips): Getting hybrid ips with creds #{@creds.inspect}")
       public_ips, private_ips = HelperFunctions.get_hybrid_ips(@creds)
     else
-      Djinn.log_debug("Getting cloud ips for #{infrastructure} with keyname #{keyname}")
+      Djinn.log_debug("Djinn (set_uaserver_ips): Getting cloud ips for #{infrastructure} with keyname #{keyname}")
       public_ips, private_ips = HelperFunctions.get_cloud_ips(infrastructure, keyname)
     end
 
-    Djinn.log_debug("Public ips are #{public_ips.join(', ')}")
-    Djinn.log_debug("Private ips are #{private_ips.join(', ')}")
-    Djinn.log_debug("Looking for #{private_ip}")
+    Djinn.log_debug("Djinn (set_uaserver_ips): Public ips are #{public_ips.join(', ')}")
+    Djinn.log_debug("Djinn (set_uaserver_ips): Private ips are #{private_ips.join(', ')}")
+    Djinn.log_debug("Djinn (set_uaserver_ips): Looking for #{private_ip}")
 
     public_ips.each_index { |index|
       node_private_ip = HelperFunctions.convert_fqdn_to_ip(private_ips[index])
@@ -874,7 +876,7 @@
     neptune_job_data=NeptuneJobData)
 
     if !file.exists?(file_to_load)
-      djinn.log_debug("No neptune data found - no need to restore")
+      djinn.log_debug("Djinn (load_neptune_info) No neptune data found - no need to restore")
       return
     end
 
@@ -962,11 +964,11 @@
       response = Net::HTTP.get_response(URI.parse(repo_url))
       data = JSON.load(response.body)
     rescue Exception => e
-      Djinn.log_debug("Update API status saw exception #{e.class}")
+      Djinn.log_debug("EXCEPTION for update_api_status connecting to the repo localhost:8079: #{e.class}. Check to see if the repo is up.")
       data = {}
     end
 
-    Djinn.log_debug("Data received is #{data.inspect}")
+    Djinn.log_debug("Djinn (update_api_status) Data received for API status is #{data.inspect}")
 
     majorities = {}
 
@@ -1228,8 +1230,10 @@
       HelperFunctions.sleep_until_port_is_open(HelperFunctions.local_ip, UA_SERVER_PORT)
     end
 
+
     start_blobstore_server if my_node.is_appengine?
 
+    start_rabbitmq_server
     # for neptune jobs, start a place where they can save output to
     # also, since repo does health checks on the app engine apis, start it up there too
 
@@ -1259,6 +1263,17 @@
     BlobServer.is_running(db_local_ip)
   end
 
+  def start_rabbitmq_server
+    # The master is run on the shadow, and all slaves are on 
+    # app engine nodes
+    if my_node.is_shadow?
+      # The secret provides a unique identifier for rabbitmq
+      RabbitMQ.start_master()
+    elsif my_node.is_appengine? or my_node.is_rabbitmq_slave?
+      # All slaves connect to the master to start
+      RabbitMQ.start_slave(get_shadow.private_ip)
+    end
+  end 
 
   def start_soap_server
     db_master_ip = nil
 
=== modified file 'AppController/lib/djinn_job_data.rb'
--- AppController/lib/djinn_job_data.rb	2012-02-01 19:17:26 +0000
+++ AppController/lib/djinn_job_data.rb	2012-02-16 06:02:21 +0000
@@ -39,8 +39,10 @@
     appscale_jobs += ["login"]
     appscale_jobs += ["memcache"]
     appscale_jobs += ["open"]
+    appscale_jobs += ["rabbitmq_slave"]
     appscale_jobs += ["babel_master", "babel_slave"]
     appscale_jobs += ["appengine"] # appengine must go last
+    
         
     appscale_jobs.each { |job|
       @jobs << job if roles.include?(job)
@@ -67,7 +69,7 @@
   def serialize
     keyname = @ssh_key.split("/")[-1]
     serialized = "#{@public_ip}:#{@private_ip}:#{@jobs.join(':')}:#{@instance_id}:#{@cloud}:#{keyname}"
-    Djinn.log_debug("serialized myself to #{serialized}")
+    Djinn.log_debug("Serialized current node to #{serialized}")
     return serialized
   end
 
@@ -76,7 +78,7 @@
     split_data = serialized.split(":")
     roles = split_data[0..-2].join(":")
     keyname = split_data[-1].split(".")[0]
-    Djinn.log_debug("i'm pretty sure roles is [#{roles}] and keyname is [#{keyname}]")
+    Djinn.log_debug("Current roles are [#{roles}] and keyname is [#{keyname}]")
     return DjinnJobData.new(roles, keyname)
   end
 
@@ -86,6 +88,7 @@
     else
       jobs = @jobs.join(', ')
     end
+
     
     status = "Node in cloud #{@cloud} with instance id #{@instance_id}" +
       " responds to ssh key #{@ssh_key}, has pub IP #{@public_ip}," +
 
=== modified file 'AppController/lib/haproxy.rb'
--- AppController/lib/haproxy.rb	2011-05-16 07:56:58 +0000
+++ AppController/lib/haproxy.rb	2012-02-16 06:02:21 +0000
@@ -209,15 +209,15 @@
   # any Mongrel, not just the one that started the session
   option redispatch
 
-  # Timeout a request if the client did not read any data for 120 seconds
-  timeout client 30000
+  # Timeout a request if the client did not read any data for 60 seconds
+  timeout client 60000
 
-  # Timeout a request if Mongrel does not accept a connection for 30 seconds
-  timeout connect 30000
+  # Timeout a request if Mongrel does not accept a connection for 60 seconds
+  timeout connect 60000
 
   # Timeout a request if Mongrel does not accept the data on the connection,
-  # or does not send a response back in 120 seconds
-  timeout server 30000
+  # or does not send a response back in 60 seconds
+  timeout server 60000
   
   # Enable the statistics page 
   stats enable
 
=== modified file 'AppController/lib/nginx.rb'
--- AppController/lib/nginx.rb	2011-06-23 01:38:14 +0000
+++ AppController/lib/nginx.rb	2012-02-16 06:02:21 +0000
@@ -89,9 +89,9 @@
       proxy_redirect off;
       proxy_pass http://gae_#{app_name};
       client_max_body_size 2G;
-      proxy_connect_timeout 30;
-      client_body_timeout 30;
-      proxy_read_timeout 30;
+      proxy_connect_timeout 60;
+      client_body_timeout 60;
+      proxy_read_timeout 60;
     }
 
     location /404.html {
@@ -178,9 +178,9 @@
       proxy_redirect off;
       proxy_pass http://#{PbServer.name};
       client_max_body_size 30M;
-      proxy_connect_timeout 30;
-      client_body_timeout 30;
-      proxy_read_timeout 30;
+      proxy_connect_timeout 60;
+      client_body_timeout 60;
+      proxy_read_timeout 60;
     }
 
 }
@@ -322,7 +322,7 @@
     #tcp_nopush     on;
 
     #keepalive_timeout  0;
-    keepalive_timeout  30;
+    keepalive_timeout  60;
     tcp_nodelay        on;
     server_names_hash_bucket_size 128;
 
 
=== modified file 'AppController/lib/pbserver.rb'
--- AppController/lib/pbserver.rb	2011-05-19 03:07:09 +0000
+++ AppController/lib/pbserver.rb	2012-02-16 06:02:21 +0000
@@ -17,7 +17,7 @@
   LISTEN_PORT = 8888
   LISTEN_SSL_PORT = 8443
   DBS_NEEDING_ONE_PBSERVER = ["mysql"]
-  DBS_WITH_NATIVE_TRANSACTIONS = ["mysql"]
+  DBS_WITH_NATIVE_PBSERVER = ["mysql"]
 
   def self.start(master_ip, db_local_ip, my_ip, table, zklocations)
     pbserver = self.pb_script(table)
@@ -87,8 +87,8 @@
   end 
 
   def self.pb_script(table)
-    if DBS_WITH_NATIVE_TRANSACTIONS.include?(table)
-      return "#{APPSCALE_HOME}/AppDB/appscale_server_native_trans.py"
+    if DBS_WITH_NATIVE_PBSERVER.include?(table)
+      return "#{APPSCALE_HOME}/AppDB/appscale_server_#{table}.py"
     else
       return "#{APPSCALE_HOME}/AppDB/appscale_server.py"
     end
 
=== added file 'AppController/lib/rabbitmq.rb'
--- AppController/lib/rabbitmq.rb	1970-01-01 00:00:00 +0000
+++ AppController/lib/rabbitmq.rb	2012-02-16 06:02:21 +0000
@@ -0,0 +1,58 @@
+#!/usr/bin/ruby -w
+
+$:.unshift File.join(File.dirname(__FILE__))
+require 'djinn_job_data'
+require 'helperfunctions'
+RABBITMQ_SERVER_PORT = 5672 
+# A class to wrap all the interactions with the ejabberd xmpp server
+class RabbitMQ
+  RABBIT_PATH = File.join("/", "etc", "rabbitmq-server")
+
+  def self.start_master()
+    Djinn.log_debug("Starting Rabbit Master")
+    set_cookie()
+    clean_start
+    start_cmd = "rabbitmq-server -detached -setcookie #{HelperFunctions.get_secret()} && rabbitmqctl reset;"
+    stop_cmd = "rabbitmqctl stop"
+
+    Djinn.log_debug(`#{start_cmd}`)
+  end
+
+  # Master IP is who to join
+  def self.start_slave(master_ip)
+    Djinn.log_debug("Starting Rabbit Slave")
+    set_cookie()
+    clean_start
+    #wait_till_port on master node
+    HelperFunctions.sleep_until_port_is_open("appscale-image0", RABBITMQ_SERVER_PORT)
+    # start the server, reset it to join the head node
+    
+    start_cmd = ["rabbitmq-server -detached -setcookie #{HelperFunctions.get_secret()}",
+                 "rabbitmqctl start_app",
+                 "rabbitmqctl stop_app",
+                 "rabbitmqctl reset", # this resets the node
+                 "rabbitmqctl cluster rabbit@appscale-image0",
+                 "rabbitmqctl start_app"]
+    start_cmd = "#{start_cmd.join('; ')}"
+    stop_cmd = "rabbitmqctl stop"
+
+    Djinn.log_debug(`#{start_cmd}`)
+    HelperFunctions.sleep_until_port_is_open("localhost", RABBITMQ_SERVER_PORT)
+  end
+
+
+  def self.stop
+    Djinn.log_debug("Shutting down rabbitmq")
+  end
+
+  def self.set_cookie()
+    cookie_file = "/var/lib/rabbitmq/.erlang.cookie"
+    File.open(cookie_file, 'w') {|f| f.write(HelperFunctions.get_secret()) }
+  end
+
+  def self.clean_start
+    Djinn.log_debug(`rm -rf /var/log/rabbitmq/*`)
+    Djinn.log_debug(`rm -rf /var/lib/rabbitmq/mnesia/*`)
+  end
+    
+end
 
=== modified file 'AppController/terminate.rb'
--- AppController/terminate.rb	2011-05-26 20:35:46 +0000
+++ AppController/terminate.rb	2012-02-16 06:02:21 +0000
@@ -135,6 +135,7 @@
  "beam", "epmd",
  # Voldemort
  "VoldemortServer",
+ "rabbitmq",
 # these are too general to kill
 # "java", "python", "python2.6", "python2.5",
  "thin", "god", "djinn", "xmpp_receiver"
 
=== modified file 'AppDB/appscale_server.py'
--- AppDB/appscale_server.py	2011-09-06 00:41:46 +0000
+++ AppDB/appscale_server.py	2012-02-16 06:02:21 +0000
@@ -150,8 +150,13 @@
     self.timeTaken = 0
   def run(self):
     s = time.time()
-    self.err, self.ret = self.db.put_entity(self.table, self.key,
+    ret = self.db.put_entity(self.table, self.key,
                                        self.fields, self.values)
+    if len(ret) > 1:
+      self.err, self.ret = ret
+    else:
+      self.err = ret[0]
+
     self.timeTaken = time.time() - s
 
 
@@ -431,7 +436,7 @@
       print "errcode:",errcode
       print "errdetail:",errdetail
     self.write( apiresponse.Encode() )
-
+    del apiresponse
 
   def _getGlobalStat(self):
     global_stat_entity=datastore.Entity("__Stat_Total__", id=1)
@@ -764,6 +769,7 @@
     clone_qr_pb.clear_cursor()
     clone_qr_pb.set_more_results( len(results)>0 )
     #logger.debug("QUERY_RESULT: %s" % clone_qr_pb)
+    del results
     return (clone_qr_pb.Encode(), 0, "")
 
 
@@ -1205,7 +1211,6 @@
       putresp_pb.key_list().append(e.key())
 
     if PROFILE: appscale_log.write("TOTAL %d %f\n"%(txn.handle(), time.time() - start))
-    
     return (putresp_pb.Encode(), 0, "")
 
 
 
=== renamed file 'AppDB/appscale_server_native_trans.py' => 'AppDB/appscale_server_mysql.py'
--- AppDB/appscale_server_native_trans.py	2011-09-09 23:51:51 +0000
+++ AppDB/appscale_server_mysql.py	2012-02-16 06:02:21 +0000
@@ -58,6 +58,7 @@
 SECRET_LOCATION = "/etc/appscale/secret.key"
 VALID_DATASTORES = []
 ERROR_CODES = []
+ID_KEY_LENGTH = 64
 app_datastore = []
 logOn = False
 logFilePtr = ""
@@ -442,6 +443,9 @@
       indexes = datastore_pb.CompositeIndices(index_proto)
       for index in indexes.index_list():
         index_map.setdefault(index.definition().entity_type(), []).append(index)
+    # TODO(nchohan)
+    # Looks like we are not storing index info in self.__indexes
+    # When GetIndicies are called it should return the indexes for said app
     self.__connection_lock.release() 
   def Clear(self):
     pass
@@ -937,14 +941,59 @@
     cursor.execute("SELECT RELEASE_LOCK('%s');" % lock_str)
     self.__connection.commit()
 
+  def __getRootKey(app_id, ancestor_list):
+    key = app_id # mysql cannot have \ as the first char in the row key
+    a = ancestor_list[0]
+    key += "/"
+
+    # append _ if the name is a number, prevents collisions of key names
+    if a.has_type():
+      key += a.type()
+    else:
+      return None
+
+    if a.has_id():
+      zero_padded_id = ("0" * (ID_KEY_LENGTH - len(str(a.id())))) + str(a.id())
+      key += ":" + zero_padded_id
+    elif a.has_name():
+      if a.name().isdigit():
+        key += ":__key__" + a.name()
+      else:
+        key += ":" + a.name()
+    else:
+      return None
+
+    return key
+
   @staticmethod
-  def __ExtractEntityGroupFromKeys(keys):
+  def __ExtractEntityGroupFromKeys(app_id, keys):
     """Extracts entity group."""
-
-    types = set([k.path().element_list()[-1].type() for k in keys])
-    assert len(types) == 1
-
-    return types.pop()
+    path = keys[0].path()
+    element_list = path.element_list()
+    def __getRootKey(app_id, ancestor_list):
+      key = app_id # mysql cannot have \ as the first char in the row key
+      a = ancestor_list[0]
+      key += "/"
+
+      # append _ if the name is a number, prevents collisions of key names
+      if a.has_type():
+        key += a.type()
+      else:
+        return None
+
+      if a.has_id():
+        zero_padded_id = ("0" * (ID_KEY_LENGTH - len(str(a.id())))) + str(a.id())
+        key += ":" + zero_padded_id
+      elif a.has_name():
+        if a.name().isdigit():
+          key += ":__key__" + a.name()
+        else:
+          key += ":" + a.name()
+      else:
+        return None
+
+      return key
+    return __getRootKey(app_id, element_list)   
 
   def AssertPbIsInitialized(self, pb):
     """Raises an exception if the given PB is not initialized and valid."""
@@ -972,9 +1021,9 @@
       entities = put_request.entity_list()
       keys = [e.key() for e in entities]
       if put_request.has_transaction():
-          entity_group = self.__ExtractEntityGroupFromKeys(keys)
-          txn_id = put_request.transaction().handle()
-          self.__AcquireLockForEntityGroup(app_id, conn, txn_id, entity_group)
+        entity_group = self.__ExtractEntityGroupFromKeys(app_id, keys)
+        txn_id = put_request.transaction().handle()
+        self.__AcquireLockForEntityGroup(app_id, conn, txn_id, entity_group)
       for entity in entities:
         self.__ValidateKey(entity.key())
 
@@ -1004,6 +1053,8 @@
 
       self.__PutEntities(conn, entities)
       put_response.key_list().extend([e.key() for e in entities])
+    except Exception, e:
+      print str(e)
     finally:
       if not put_request.has_transaction():
         self.__ReleaseConnection(conn)
@@ -1013,7 +1064,7 @@
     try:
       keys = get_request.key_list()
       if get_request.has_transaction():
-          entity_group = self.__ExtractEntityGroupFromKeys(keys)
+          entity_group = self.__ExtractEntityGroupFromKeys(app_id, keys)
           txn_id = get_request.transaction().handle()
           self.__AcquireLockForEntityGroup(app_id, conn, txn_id, entity_group)
       for key in keys:
@@ -1036,7 +1087,7 @@
     try:
       keys = delete_request.key_list()
       if delete_request.has_transaction():
-        entity_group = self.__ExtractEntityGroupFromKeys(keys)
+        entity_group = self.__ExtractEntityGroupFromKeys(app_id, keys)
         txn_id = delete_request.transaction().handle()
         self.__AcquireLockForEntityGroup(app_id, conn, txn_id, entity_group)
       self.__DeleteEntities(conn, delete_request.key_list())
@@ -1525,11 +1576,18 @@
                                              'New index id must be 0.')
 
     self.__index_lock.acquire()
+
+    # If it already exists, just return the index id
+    if self.__FindIndex(index):
+      self.__index_lock.release()
+      id_response.set_value(self.__FindIndex(index))
+      return 
+
     try:
-      if self.__FindIndex(index):
-        raise apiproxy_errors.ApplicationError(datastore_pb.Error.BAD_REQUEST,
-                                               'Index already exists.')
-
+        #raise apiproxy_errors.ApplicationError(
+        #                          datastore_pb.Error.PERMISSION_DENIED,
+        #                          'Index already exists.')
+      
       next_id = max([idx.id() for x in self.__indexes.get(app_id, {}).values()
                      for idx in x] + [0]) + 1
       index.set_id(next_id)
@@ -1544,6 +1602,8 @@
         self.__WriteIndexData(conn, app_id)
       finally:
         self.__ReleaseConnection(conn)
+    except Exception, e:
+      print str(e)
     finally:
       self.__index_lock.release()
 
@@ -1631,7 +1691,6 @@
       apirequest.clear_request()
     method = apirequest.method()
     http_request_data = apirequest.request()
-
     if method == "Put":
       response, errcode, errdetail = self.put_request(app_id, 
                                                  http_request_data)
@@ -1687,14 +1746,16 @@
     self.write(apiresponse.Encode() )    
 
   def create_index(self, app_id, http_request_data):
-    index = entity_pb.Index(http_request_data)
+    index = entity_pb.CompositeIndex(http_request_data)
     integer = api_base_pb.Integer64Proto()
     try:
       app_datastore._Dynamic_CreateIndex(index, integer)
     except Exception, e:
+      print str(e)
       return (api_base_pb.VoidProto().Encode(),
               datastore_pb.Error.INTERNAL_ERROR, 
               str(e))
+    return (integer.Encode(), 0, "")
 
   def get_indices(self, app_id, http_request_data):
     composite_indices = datastore_pb.CompositeIndices()
@@ -1707,8 +1768,8 @@
     return (composite_indices.Encode(), 0, "")
 
   def update_index(self, app_id, http_request_data):
-    index = entity_pb.Index(http_request_data)
-    void_resp = api_base_pb.VoidProto().Encode()
+    index = entity_pb.CompositeIndex(http_request_data)
+    void_resp = api_base_pb.VoidProto()
     try:
       app_datastore._Dynamic_UpdateIndex(index, void_resp) 
     except Exception, e:
@@ -1776,6 +1837,7 @@
     txn_id = transaction_pb.handle() 
     commitres_pb = datastore_pb.CommitResponse()
     try:
+      zoo_keeper.releaseLock(app_id, txn_id)
       app_datastore._Dynamic_Commit(app_id, transaction_pb, commitres_pb)
     except:
       return (commitres_pb.Encode(), datastore_pb.Error.PERMISSION_DENIED, "Unable to commit for this transaction")  
@@ -1785,9 +1847,12 @@
     transaction_pb = datastore_pb.Transaction(http_request_data)
     handle = transaction_pb.handle() 
     try:
+      zoo_keeper.releaseLock(app_id, handle)
       app_datastore._Dynamic_Rollback(app_id, transaction_pb, None)
-    except:
+    except Exception, e:
+      print str(e)
       return(api_base_pb.VoidProto().Encode(), datastore_pb.Error.PERMISSION_DENIED, "Unable to rollback for this transaction")
+    print "Transaction with handle %d was roll backed"%handle
     return (api_base_pb.VoidProto().Encode(), 0, "")
 
 
 
=== modified file 'AppDB/cassandra/cassandra_helper.rb'
--- AppDB/cassandra/cassandra_helper.rb	2011-05-30 01:04:15 +0000
+++ AppDB/cassandra/cassandra_helper.rb	2012-02-16 06:02:21 +0000
@@ -15,11 +15,29 @@
   return false
 end
 
+def get_local_token(master_ip, slave_ips)
+  # Calculate everyone's token for data partitioning
+  if master_ip == HelperFunctions.local_ip
+    return 0
+  end
+    
+  for ii in 0..slave_ips.length
+    # Based on local ip return the correct token
+    # This token generation was taken from:
+    # http://www.datastax.com/docs/0.8/install/cluster_init#cluster-init
+    if slave_ips[ii] == HelperFunctions.local_ip
+      # Add one to offset the master
+      return (ii + 1)*(2**127)/(1 + slave_ips.length)
+    end
+  end
+end
+
 def setup_db_config_files(master_ip, slave_ips, creds)
   source_dir = "#{APPSCALE_HOME}/AppDB/cassandra/templates"
   dest_dir = "#{APPSCALE_HOME}/AppDB/cassandra/cassandra/conf"
 
   all_ips = [master_ip, slave_ips].flatten
+  local_token = get_local_token(master_ip, slave_ips)
 
   files_to_config = `ls #{source_dir}`.split
   files_to_config.each{ |filename|
@@ -29,6 +47,7 @@
       contents = source_file.read
       contents.gsub!(/APPSCALE-LOCAL/, HelperFunctions.local_ip)
       contents.gsub!(/APPSCALE-MASTER/, master_ip)
+      contents.gsub!(/APPSCALE-TOKEN/, "#{local_token}")
       contents.gsub!(/REPLICATION/, creds["replication"])
       contents.gsub!(/APPSCALE-JMX-PORT/, "7070")              
       File.open(full_path_to_write, "w+") { |dest_file|
@@ -36,16 +55,16 @@
       }
     }
   }
+  
 end
-
 def start_db_master()
   @state = "Starting up Cassandra on the head node"
   Djinn.log_debug("Starting up Cassandra as master")
 
-  Djinn.log_debug(`pkill ThriftBroker`)
+  Djinn.log_run("pkill ThriftBroker")
   `rm -rf /var/appscale/cassandra*`
-
-  Djinn.log_debug(`#{APPSCALE_HOME}/AppDB/cassandra/cassandra/bin/cassandra start -p /var/appscale/appscale-cassandra.pid`)
+  `rm /var/log/appscale/cassandra/system.log`
+  Djinn.log_run("#{APPSCALE_HOME}/AppDB/cassandra/cassandra/bin/cassandra start -p /var/appscale/appscale-cassandra.pid")
   HelperFunctions.sleep_until_port_is_open(HelperFunctions.local_ip, 9160)
 end
 
@@ -54,10 +73,12 @@
   Djinn.log_debug("Starting up Cassandra as slave")
 
   HelperFunctions.sleep_until_port_is_open(Djinn.get_db_master_ip, 9160)
-    
+  sleep(5)
   `rm -rf /var/appscale/cassandra*`
-  Djinn.log_debug(`#{APPSCALE_HOME}/AppDB/cassandra/cassandra/bin/cassandra start -p /var/appscale/appscale-cassandra.pid`)
-  HelperFunctions.sleep_until_port_is_open(Djinn.get_db_master_ip, 9160)
+  `rm /var/log/appscale/cassandra/system.log`
+  `#{APPSCALE_HOME}/AppDB/cassandra/cassandra/bin/cassandra start -p /var/appscale/appscale-cassandra.pid`
+  #Djinn.log_run("#{APPSCALE_HOME}/AppDB/cassandra/cassandra/bin/cassandra start -p /var/appscale/appscale-cassandra.pid")
+  HelperFunctions.sleep_until_port_is_open(HelperFunctions.local_ip, 9160)
 end
 
 def stop_db_master
@@ -70,3 +91,4 @@
   Djinn.log_run("#{APPSCALE_HOME}/AppDB/cassandra/cassandra/bin/nodetool decommission -h #{HelperFunctions.local_ip} -p 6666")
   Djinn.log_run("cat /var/appscale/appscale-cassandra.pid | xargs kill -9")
 end
+
 
=== modified file 'AppDB/cassandra/prime_cassandra.py'
--- AppDB/cassandra/prime_cassandra.py	2011-05-29 21:41:44 +0000
+++ AppDB/cassandra/prime_cassandra.py	2012-02-16 06:02:21 +0000
@@ -18,19 +18,19 @@
   except pycassa.cassandra.ttypes.InvalidRequestException, e:
     pass
 
-  sys.create_keyspace('Keyspace1', replication)
-  sys.create_column_family('Keyspace1', 'Standard1', 
-                          comparator_type=UTF8_TYPE)
-  sys.create_column_family('Keyspace1', 'Standard2',
-                          comparator_type=UTF8_TYPE)
-  sys.create_column_family('Keyspace1', 'StandardByTime1',
-                          comparator_type=TIME_UUID_TYPE)
-  sys.create_column_family('Keyspace1', 'StandardByTime2',
-                          comparator_type=TIME_UUID_TYPE)
-  sys.create_column_family('Keyspace1', 'Super1', super=True,
-                          comparator_type=UTF8_TYPE)
-  sys.create_column_family('Keyspace1', 'Super2', super=True,
-                          comparator_type=UTF8_TYPE)
+  sys.create_keyspace('Keyspace1', pycassa.SIMPLE_STRATEGY, {'replication_factor':str(replication)})
+  sys.create_column_family('Keyspace1', 'Standard1', #column_type="Standard",
+                          comparator_type=UTF8_TYPE)
+  sys.create_column_family('Keyspace1', 'Standard2', #column_type="Standard",
+                          comparator_type=UTF8_TYPE)
+  sys.create_column_family('Keyspace1', 'StandardByTime1', #column_type="Standard",
+                          comparator_type=TIME_UUID_TYPE)
+  sys.create_column_family('Keyspace1', 'StandardByTime2', #column_type="Standard",
+                          comparator_type=TIME_UUID_TYPE)
+  #sys.create_column_family('Keyspace1', 'Super1',  column_type="Super",
+  #                        comparator_type=UTF8_TYPE)
+  #sys.create_column_family('Keyspace1', 'Super2', column_type="Super",
+  #                        comparator_type=UTF8_TYPE)
   sys.close()
   print "SUCCESS"
 
@@ -41,7 +41,6 @@
   #print db.get("__keys_") 
   db.create_table(USERS_TABLE, USERS_SCHEMA)
   db.create_table(APPS_TABLE, APPS_SCHEMA)
-
   if len(db.get_schema(USERS_TABLE)) > 1 and len(db.get_schema(APPS_TABLE)) > 1:
     print "CREATE TABLE SUCCESS FOR USER AND APPS"
     print db.get_schema(USERS_TABLE)
 
=== modified file 'AppDB/cassandra/py_cassandra.py'
--- AppDB/cassandra/py_cassandra.py	2011-05-31 00:01:41 +0000
+++ AppDB/cassandra/py_cassandra.py	2012-02-16 06:02:21 +0000
@@ -1,36 +1,31 @@
 #
 # Cassandra Interface for AppScale
-# Rewritten by Navraj Chohan for using range queries
-# Modified by Chris Bunch for upgrade to Cassandra 0.50.0
-# on 2/17/10
+# Rewritten by Navraj Chohan for pycassa
+# Modified by Chris Bunch for upgrade to Cassandra 0.50.0 # on 2/17/10
 # Original author: [e-mail]
 
 import os,sys
 import time
-
-from thrift_cass.Cassandra import Client
-from thrift_cass.ttypes import *
-
 import string
-import base64   # base64    2009.04.16
+import base64   
 from dbconstants import *
 from dbinterface import *
-#import sqlalchemy.pool as pool
 import appscale_logger
 import pycassa
+from pycassa.system_manager import *
 from thrift import Thrift
 from thrift.transport import TSocket
 from thrift.transport import TTransport
 from thrift.protocol import TBinaryProtocol
+from pycassa.cassandra.ttypes import NotFoundException
+
 ERROR_DEFAULT = "DB_ERROR:" # ERROR_CASSANDRA
 # Store all schema information in a special table
 # If a table does not show up in this table, try a range query 
 # to discover it's schema
 SCHEMA_TABLE = "__key__"
 SCHEMA_TABLE_SCHEMA = ['schema']
-# use 1 Table and 1 ColumnFamily in Cassandra
 MAIN_TABLE = "Keyspace1"
-COLUMN_FAMILY = "Standard1"
 
 PERSISTENT_CONNECTION = False
 PROFILING = False
@@ -50,9 +45,20 @@
     f = open(APPSCALE_HOME + '/.appscale/my_private_ip', 'r')
     self.host = f.read()
     self.port = DEFAULT_PORT
-    #self.pool = pool.QueuePool(self.__create_connection, reset_on_return=False)
-    #connection.add_pool('AppScale', [self.host, self.port])
-    self.pool = pycassa.ConnectionPool(keyspace='Keyspace1', server_list=[self.host+":"+str(self.port)], prefill=False)
+    self.pool = pycassa.ConnectionPool(keyspace='Keyspace1', 
+                           server_list=[self.host+":"+str(self.port)], 
+                           prefill=False)
+    f = open(APPSCALE_HOME + '/.appscale/my_private_ip', 'r')
+    host = f.read()
+    sys = SystemManager(host + ":" + str(DEFAULT_PORT))
+    try: 
+      sys.create_column_family('Keyspace1', 
+                               SCHEMA_TABLE, 
+                               comparator_type=UTF8_TYPE)
+    except Exception, e:
+      print "Exception creating column family: %s"%str(e)
+      pass
+
     self.logger = logger
 
   def logTiming(self, function, start_time, end_time):
@@ -62,30 +68,21 @@
   def get_entity(self, table_name, row_key, column_names):
     error = [ERROR_DEFAULT]
     list = error
-    client = None
     row_key = table_name + '/' + row_key
     try:
-      slice_predicate = SlicePredicate(column_names=column_names)
-      path = ColumnPath(COLUMN_FAMILY)
-      client = self.__setup_connection()
-      # Result is a column type which has name, value, timestamp
-      result = client.get_slice(row_key, path, slice_predicate,
-                                 CONSISTENCY_QUORUM)
+      cf = pycassa.ColumnFamily(self.pool, 
+                                string.replace(table_name, '-','a'))
+      result = cf.get(row_key, columns=column_names)
+      # Order entities by column_names 
       for column in column_names:
-        for r in result:
-          c = r.column
-          if column == c.name:
-            list.append(c.value)
-    except NotFoundException: # occurs normally if the item isn't in the db 
+        list.append(result[column])
+    except NotFoundException: 
       list[0] += "Not found"
-      self.__close_connection(client)
       return list
     except Exception, ex:
-      #self.logger.debug("Exception %s" % ex)
       list[0]+=("Exception: %s"%ex)
-      self.__close_connection(client)
       return list
-    self.__close_connection(client)
+
     if len(list) == 1:
       list[0] += "Not found"
     return list
@@ -93,34 +90,25 @@
   def put_entity(self, table_name, row_key, column_names, cell_values):
     error = [ERROR_DEFAULT]
     list = error
-    client = None
 
     # The first time a table is seen
     if table_name not in table_cache:
       self.create_table(table_name, column_names)
 
     row_key = table_name + '/' + row_key
-    client = self.__setup_connection()
-    curtime = self.timestamp()
-    # Result is a column type which has name, value, timestamp
-    mutations = []
+    cell_dict = {}
     for index, ii in enumerate(column_names):
-      column = Column(name = ii, value=cell_values[index],
-                    timestamp=curtime)
-      c_or_sc = ColumnOrSuperColumn(column=column)
-      mutation = Mutation(column_or_supercolumn=c_or_sc)
-      mutations.append(mutation)
-    mutation_map = {row_key : { COLUMN_FAMILY : mutations } }
-    client.batch_mutate(mutation_map, CONSISTENCY_QUORUM) 
-    """except Exception, ex:
-      print "EXCEPTION"
-      self.logger.debug("Exception %s" % ex)
-      list[0]+=("Exception: %s"%ex)
-      self.__close_connection(client)
-      list.append("0")
+      cell_dict[ii] = cell_values[index]
+
+    try:
+      # cannot have "-" in the column name
+      cf = pycassa.ColumnFamily(self.pool, string.replace(table_name, '-','a'))
+    except NotFoundException:
+      print "Unable to find column family"
+      list[0]+=("Exception: Column family not found")
       return list
-    """
-    self.__close_connection(client)
+
+    cf.insert(row_key, cell_dict)
     list.append("0")
     return list
 
@@ -130,18 +118,19 @@
 
   def get_table(self, table_name, column_names):
     error = [ERROR_DEFAULT]  
-    client = None
     result = error
     keyslices = []
     start_key = table_name + "/"
     end_key = table_name + '/~'
     try: 
-      cf = pycassa.ColumnFamily(self.pool, 'Standard1')
+      cf = pycassa.ColumnFamily(self.pool, string.replace(table_name, '-','a'))
       keyslices = cf.get_range(columns=column_names, 
                               start=start_key, 
-                              finish=end_key, 
-                              read_consistency_level=CONSISTENCY_QUORUM)
+                              finish=end_key)
       keyslices = list(keyslices)
+    except pycassa.NotFoundException, ex:
+      self.logger.debug("No column fam yet--exception %s" % ex)
+      return result
     except Exception, ex:
       self.logger.debug("Exception %s" % ex)
       result[0] += "Exception: " + str(ex)
@@ -166,21 +155,15 @@
   def delete_row(self, table_name, row_key):
     error = [ERROR_DEFAULT]
     ret = error
-    client = None
     row_key = table_name + '/' + row_key
-    path = ColumnPath(COLUMN_FAMILY)
     try: 
-      client = self.__setup_connection()
-      curtime = self.timestamp()
+      cf = pycassa.ColumnFamily(self.pool, string.replace(table_name, '-','a'))
       # Result is a column type which has name, value, timestamp
-      client.remove(row_key, path, curtime,
-                               CONSISTENCY_QUORUM) 
+      cf.remove(row_key)
     except Exception, ex:
       self.logger.debug("Exception %s" % ex)
       ret[0]+=("Exception: %s"%ex)
-      self.__close_connection(client)
       return ret 
-    self.__close_connection(client)
     ret.append("0")
     return ret
 
@@ -203,51 +186,47 @@
   def delete_table(self, table_name):
     error = [ERROR_DEFAULT]  
     result = error
-    client = None
     keyslices = []
-    column_parent = ColumnParent(column_family="Standard1")
-    predicate = SlicePredicate(column_names=[])
     curtime = self.timestamp()
-    path = ColumnPath(COLUMN_FAMILY)
     start_key = table_name + "/"
     end_key = table_name + '/~'
     try: 
-      cf = pycassa.ColumnFamily(self.pool, 'Standard1')
-      keyslices = cf.get_range(columns=[], 
-                              start=start_key, 
-                              finish=end_key, 
-                              read_consistency_level=CONSISTENCY_QUORUM)
+      cf = pycassa.ColumnFamily(self.pool, string.replace(table_name, '-','a'))
+      cf.truncate()
+      self.delete_row(SCHEMA_TABLE, row_key)
     except Exception, ex:
       self.logger.debug("Exception %s" % ex)
       result[0]+=("Exception: %s"%ex)
       return result
-    keys_removed = False
-    for keyslice in keyslices:
-      row_key = keyslice[0]
-      client = self.__setup_connection()
-      client.remove(row_key, 
-                    path, 
-                    curtime,
-                    CONSISTENCY_QUORUM) 
-      keys_removed = True
-    if table_name not in table_cache and keys_removed:
+    if table_name not in table_cache:
       result[0] += "Table does not exist"
       return  result
     if table_name in table_cache:
       del table_cache[table_name]
-    if client: 
-      self.__close_connection(client)
     return result
 
   # Only stores the schema
   def create_table(self, table_name, column_names):
-    table_cache[table_name] = 1
     columns = ':'.join(column_names)
     row_key = table_name
+    print "CREATE TABLE NAME: " + table_name
     # Get and make sure we are not overwriting previous schemas
     ret = self.get_entity(SCHEMA_TABLE, row_key, SCHEMA_TABLE_SCHEMA)
+    print ret
     if ret[0] != ERROR_DEFAULT:
-      self.put_entity(SCHEMA_TABLE, row_key, SCHEMA_TABLE_SCHEMA, [columns])
+      f = open(APPSCALE_HOME + '/.appscale/my_private_ip', 'r')
+      host = f.read()
+      sysman = SystemManager(host + ":" + str(DEFAULT_PORT))
+      print "Creating column family %s"%table_name
+      try:
+        sysman.create_column_family('Keyspace1', string.replace(table_name, '-','a'), comparator_type=UTF8_TYPE)
+        print "Done creating column family"
+        self.put_entity(SCHEMA_TABLE, row_key, SCHEMA_TABLE_SCHEMA, [columns])
+      except Exception, e:
+        print "Unable to create column family %s"%str(e)
+        return
+
+    table_cache[table_name] = 1
 
   ######################################################################
   # private methods 
 
=== added file 'AppDB/cassandra/templates/brisk'
--- AppDB/cassandra/templates/brisk	1970-01-01 00:00:00 +0000
+++ AppDB/cassandra/templates/brisk	2012-02-16 06:02:21 +0000
@@ -0,0 +1,16 @@
+# NOTICE: See also /etc/brisk/cassandra/cassandra-env.sh
+
+# EXTRA_CLASSPATH provides the means to extend Cassandra's classpath with
+# additional libraries.  It is formatted as a colon-delimited list of
+# class directories and/or jar files.  For example, to enable the
+# JMX-to-web bridge install libmx4j-java and uncomment the following.
+#EXTRA_CLASSPATH="/usr/share/java/mx4j-tools.jar"
+
+# enable this start also start Hadoop's JobTracker and/or TaskTrackers on this
+# machine.  If left disabled, this will act as a regular Cassandra node.
+HADOOP_ENABLED=1
+
+# enable this to set the replication factor for CFS.  Note that this will only
+# have an effect the first time a cluster is started with HADOOP_ENABLED=1 and
+# after that will be a no-op. Defaults to 1.
+#CFS_REPLICATION_FACTOR=1
 
=== modified file 'AppDB/cassandra/templates/cassandra.yaml'
--- AppDB/cassandra/templates/cassandra.yaml	2011-05-30 01:04:15 +0000
+++ AppDB/cassandra/templates/cassandra.yaml	2012-02-16 06:02:21 +0000
@@ -19,23 +19,15 @@
 # the heaviest-loaded existing node.  If there is no load information
 # available, such as is the case with a new cluster, it will pick
 # a random token, which will lead to hot spots.
-initial_token:
-
-# Set to true to make new [non-seed] nodes automatically migrate data
-# to themselves from the pre-existing nodes in the cluster.  Defaults
-# to false because you can only bootstrap N machines at a time from
-# an existing cluster of N, so if you are bringing up a cluster of
-# 10 machines with 3 seeds you would have to do it in stages.  Leaving
-# this off for the initial start simplifies that.
-auto_bootstrap: false
+initial_token: APPSCALE-TOKEN
 
 # See http://wiki.apache.org/cassandra/HintedHandoff
 hinted_handoff_enabled: true
 # this defines the maximum amount of time a dead host will have hints
 # generated.  After it has been dead this long, hints will be dropped.
 max_hint_window_in_ms: 3600000 # one hour
-# Sleep this long after delivering each row or row fragment
-hinted_handoff_throttle_delay_in_ms: 50
+# Sleep this long after delivering each hint
+hinted_handoff_throttle_delay_in_ms: 1
 
 # authentication backend, implementing IAuthenticator; used to identify users
 authenticator: org.apache.cassandra.auth.AllowAllAuthenticator
@@ -64,10 +56,10 @@
 #   ordering.  Use this as an example if you need custom collation.
 #
 # See http://wiki.apache.org/cassandra/Operations for more on
-# partitioners anv toke liblib libselection.
+# partitioners and token selection.
 partitioner: org.apache.cassandra.dht.ByteOrderedPartitioner
 
-# directories where Cassandra should libstore data on disk.
+# directories where Cassandra should store data on disk.
 data_file_directories:
     - /var/appscale/cassandra/data
 
@@ -77,21 +69,34 @@
 # saved caches
 saved_caches_directory: /var/appscale/cassandra/saved_caches
 
-# Size to allow commitlog to grow to before creating a new segment 
-commitlog_rotation_threshold_in_mb: 128
-
 # commitlog_sync may be either "periodic" or "batch." 
 # When in batch mode, Cassandra won't ack writes until the commit log
 # has been fsynced to disk.  It will wait up to
-# CommitLogSyncBatchWindowInMS milliseconds for other writes, before
+# commitlog_sync_batch_window_in_ms milliseconds for other writes, before
 # performing the sync.
-commitlog_sync: periodic
-
+#
+# commitlog_sync: batch
+# commitlog_sync_batch_window_in_ms: 50
+#
 # the other option is "periodic" where writes may be acked immediately
 # and the CommitLog is simply synced every commitlog_sync_period_in_ms
 # milliseconds.
+commitlog_sync: periodic
 commitlog_sync_period_in_ms: 10000
 
+# any class that implements the SeedProvider interface and has a
+# constructor that takes a Map<String, String> of parameters will do.
+seed_provider:
+    # Addresses of hosts that are deemed contact points. 
+    # Cassandra nodes use this list of hosts to find each other and learn
+    # the topology of the ring.  You must change this if you are running
+    # multiple nodes!
+    - class_name: org.apache.cassandra.locator.SimpleSeedProvider
+      parameters:
+          # seeds is actually a comma-delimited list of addresses.
+          # Ex: "<ip1>,<ip2>,<ip3>"
+          - seeds: "APPSCALE-MASTER"
+
 # emergency pressure valve: each time heap usage after a full (CMS)
 # garbage collection is above this fraction of the max, Cassandra will
 # flush the largest memtables.  
@@ -117,13 +122,6 @@
 reduce_cache_sizes_at: 0.85
 reduce_cache_capacity_to: 0.6
 
-# Addresses of hosts that are deemed contact points.
-# Cassandra nodes use this list of hosts to find each other and learn
-# the topology of the ring.  You must change this if you are running
-# multiple nodes!
-seeds:
- - APPSCALE-MASTER
-
 # For workloads with more data than can fit in memory, Cassandra's
 # bottleneck will be reads that need to fetch data from
 # disk. "concurrent_reads" should be set to (16 * number_of_drives) in
@@ -136,6 +134,17 @@
 concurrent_reads: 32
 concurrent_writes: 32
 
+# Total memory to use for memtables.  Cassandra will flush the largest
+# memtable when this much memory is used.
+# If omitted, Cassandra will set it to 1/3 of the heap.
+# memtable_total_space_in_mb: 2048
+
+# Total space to use for commitlogs. 
+# If space gets above this value (it will round up to the next nearest
+# segment multiple), Cassandra will flush every dirty CF in the oldest
+# segment and remove it.
+# commitlog_total_space_in_mb: 4096
+
 # This sets the amount of memtable flush writer threads.  These will
 # be blocked by disk io, and each one will hold a memtable in memory
 # while blocked. If you have a large heap and many data directories,
@@ -155,6 +164,10 @@
 # TCP port, for commands and data
 storage_port: 7000
 
+# SSL port, for encrypted communication.  Unused unless enabled in
+# encryption_options
+ssl_storage_port: 7001
+
 # Address to bind to and tell other Cassandra nodes to connect to. You
 # _must_ change this if you want multiple nodes to be able to
 # communicate!
@@ -167,29 +180,53 @@
 # Setting this to 0.0.0.0 is always wrong.
 listen_address: APPSCALE-LOCAL
 
+# Address to broadcast to other Cassandra nodes
+# Leaving this blank will set it to the same value as listen_address
+# broadcast_address: 1.2.3.4
+
 # The address to bind the Thrift RPC service to -- clients connect
 # here. Unlike ListenAddress above, you *can* specify 0.0.0.0 here if
 # you want Thrift to listen on all interfaces.
 # 
 # Leaving this blank has the same effect it does for ListenAddress,
 # (i.e. it will be based on the configured hostname of the node).
-rpc_address: APPSCALE-LOCAL
+rpc_address: 0.0.0.0
 # port for Thrift to listen for clients on
 rpc_port: 9160
 
 # enable or disable keepalive on rpc connections
 rpc_keepalive: true
 
-# Cassandra uses thread-per-client for client RPC.  This can
-# be expensive in memory used for thread stack for a large
-# enough number of clients.  (Hence, connection pooling is
-# very, very strongly recommended.)
-# 
+# Cassandra provides three options for the RPC Server:
+#
+# sync  -> One connection per thread in the rpc pool (see below).
+#          For a very large number of clients, memory will be your limiting
+#          factor; on a 64 bit JVM, 128KB is the minimum stack size per thread.
+#          Connection pooling is very, very strongly recommended.
+#
+# async -> Nonblocking server implementation with one thread to serve 
+#          rpc connections.  This is not recommended for high throughput use
+#          cases. Async has been tested to be about 50% slower than sync
+#          or hsha and is deprecated: it will be removed in the next major release.
+#
+# hsha  -> Stands for "half synchronous, half asynchronous." The rpc thread pool 
+#          (see below) is used to manage requests, but the threads are multiplexed
+#          across the different clients.
+#
+# The default is sync because on Windows hsha is about 30% slower.  On Linux,
+# sync/hsha performance is about the same, with hsha of course using less memory.
+rpc_server_type: sync
+
 # Uncomment rpc_min|max|thread to set request pool size.
-# You would primarily set max as a safeguard against misbehaved
-# clients; if you do hit the max, Cassandra will block until
-# one disconnects before accepting more.  The defaults are
-# min of 16 and max unlimited.
+# You would primarily set max for the sync server to safeguard against
+# misbehaved clients; if you do hit the max, Cassandra will block until one
+# disconnects before accepting more.  The defaults for sync are min of 16 and max
+# unlimited.
+# 
+# For the Hsha server, the min and max both default to quadruple the number of
+# CPU cores.
+#
+# This configuration is ignored by the async server.
 #
 # rpc_min_threads: 16
 # rpc_max_threads: 2048
@@ -219,10 +256,6 @@
 # is a data format change.
 snapshot_before_compaction: false
 
-# change this to increase the compaction thread's priority.  In java, 1 is the
-# lowest priority and that is our default.
-# compaction_thread_priority: 1
-
 # Add column indexes to a row after its contents reach this size.
 # Increase if your column values are large, or if you have a very large
 # number of columns.  The competing causes are, Cassandra has to
@@ -237,11 +270,48 @@
 # will be logged specifying the row key.
 in_memory_compaction_limit_in_mb: 64
 
+# Number of simultaneous compactions to allow, NOT including
+# validation "compactions" for anti-entropy repair.  Simultaneous
+# compactions can help preserve read performance in a mixed read/write
+# workload, by mitigating the tendency of small sstables to accumulate
+# during a single long running compactions. The default is usually
+# fine and if you experience problems with compaction running too
+# slowly or too fast, you should look at
+# compaction_throughput_mb_per_sec first.
+#
+# This setting has no effect on LeveledCompactionStrategy.
+#
+# concurrent_compactors defaults to the number of cores.
+# Uncomment to make compaction mono-threaded, the pre-0.8 default.
+#concurrent_compactors: 1
+
+# Multi-threaded compaction. When enabled, each compaction will use
+# up to one thread per core, plus one thread per sstable being merged.
+# This is usually only useful for SSD-based hardware: otherwise, 
+# your concern is usually to get compaction to do LESS i/o (see:
+# compaction_throughput_mb_per_sec), not more.
+multithreaded_compaction: false
+
+# Throttles compaction to the given total throughput across the entire
+# system. The faster you insert data, the faster you need to compact in
+# order to keep the sstable count down, but in general, setting this to
+# 16 to 32 times the rate you are inserting data is more than sufficient.
+# Setting this to 0 disables throttling. Note that this account for all types
+# of compaction, including validation compaction.
+compaction_throughput_mb_per_sec: 16
+
 # Track cached row keys during compaction, and re-cache their new
 # positions in the compacted sstable.  Disable if you use really large
 # key caches.
 compaction_preheat_key_cache: true
 
+# Throttles all outbound streaming file transfers on this node to the
+# given total throughput in Mbps. This is necessary because Cassandra does
+# mostly sequential IO when streaming data during bootstrap or repair, which
+# can lead to saturating the network connection and degrading rpc performance.
+# When unset, the default is 400 Mbps or 50 MB/s.
+# stream_throughput_outbound_megabits_per_sec: 400
+
 # Time to wait for a reply from other nodes before failing the command 
 rpc_timeout_in_ms: 10000
 
@@ -265,11 +335,6 @@
 #    explicitly configured in cassandra-topology.properties.
 endpoint_snitch: org.apache.cassandra.locator.SimpleSnitch
 
-# dynamic_snitch -- This boolean controls whether the above snitch is
-# wrapped with a dynamic snitch, which will monitor read latencies
-# and avoid reading from hosts that have slowed (due to compaction,
-# for instance)
-dynamic_snitch: true
 # controls how often to perform the more expensive part of host score
 # calculation
 dynamic_snitch_update_interval_in_ms: 100 
@@ -283,7 +348,7 @@
 # expressed as a double which represents a percentage.  Thus, a value of
 # 0.2 means Cassandra would continue to prefer the static snitch values
 # until the pinned host was 20% worse than the fastest.
-dynamic_snitch_badness_threshold: 0.0
+dynamic_snitch_badness_threshold: 0.1
 
 # request_scheduler -- Set this to a class that implements
 # RequestScheduler, which will schedule incoming client requests
@@ -325,32 +390,40 @@
 # the request scheduling. Currently the only valid option is keyspace.
 # request_scheduler_id: keyspace
 
-# The Index Interval determines how large the sampling of row keys
-#  is for a given SSTable. The larger the sampling, the more effective
-#  the index is at the cost of space.
+# index_interval controls the sampling of entries from the primrary
+# row index in terms of space versus time.  The larger the interval,
+# the smaller and less effective the sampling will be.  In technicial
+# terms, the interval coresponds to the number of index entries that
+# are skipped between taking each sample.  All the sampled entries
+# must fit in memory.  Generally, a value between 128 and 512 here
+# coupled with a large key cache size on CFs results in the best trade
+# offs.  This value is not often changed, however if you have many
+# very small rows (many to an OS page), then increasing this will
+# often lower memory usage without a impact on performance.
 index_interval: 128
 
-
-#keyspaces:
-#- column_families:
-#  - column_type: Standard
-#    name: Standard1
-#  - column_type: Standard
-#    name: Standard2
-#  - column_type: Standard
-#    compare_with: org.apache.cassandra.db.marshal.TimeUUIDType
-#    name: StandardByTime1
-#  - column_type: Standard
-#    compare_with: org.apache.cassandra.db.marshal.TimeUUIDType
-#    name: StandardByTime2
-#  - column_type: Super
-#    name: Super1
-#  - column_type: Super
-#    name: Super2
-#    
-#  name: Keyspace1
-#  replica_placement_strategy: org.apache.cassandra.locator.SimpleStrategy
-#  replica_placement_factor: REPLICATION
-#
-#
-
+# Enable or disable inter-node encryption
+# Default settings are TLS v1, RSA 1024-bit keys (it is imperative that
+# users generate their own keys) TLS_RSA_WITH_AES_128_CBC_SHA as the cipher
+# suite for authentication, key exchange and encryption of the actual data transfers.
+# NOTE: No custom encryption options are enabled at the moment
+# The available internode options are : all, none, dc, rack
+#
+# If set to dc cassandra will encrypt the traffic between the DCs
+# If set to rack cassandra will encrypt the traffic between the racks
+#
+# The passwords used in these options must match the passwords used when generating
+# the keystore and truststore.  For instructions on generating these files, see:
+# http://download.oracle.com/javase/6/docs/technotes/guides/security/jsse/JSSERefGuide.html#CreateKeystore
+#
+encryption_options:
+    internode_encryption: none
+    keystore: conf/.keystore
+    keystore_password: cassandra
+    truststore: conf/.truststore
+    truststore_password: cassandra
+    # More advanced defaults below:
+    # protocol: TLS
+    # algorithm: SunX509
+    # store_type: JKS
+    # cipher_suites: [TLS_RSA_WITH_AES_128_CBC_SHA,TLS_RSA_WITH_AES_256_CBC_SHA]
 
=== modified file 'AppDB/cassandra/test_cassandra.py' (properties changed: +x to -x)
--- AppDB/cassandra/test_cassandra.py	2011-05-29 21:41:44 +0000
+++ AppDB/cassandra/test_cassandra.py	2012-02-16 06:02:21 +0000
@@ -1,286 +1,89 @@
-#
-# Cassandra Interface for AppScale
-# Rewritten by Navraj Chohan for using range queries
-# Modified by Chris Bunch for upgrade to Cassandra 0.50.0
-# on 2/17/10
-# Original author: [e-mail]
-
-import os,sys
-import time
-
-from thrift_cass.Cassandra import Client
-from thrift_cass.ttypes import *
-
-import string
-import base64   # base64    2009.04.16
-from dbconstants import *
-from dbinterface import *
-#import sqlalchemy.pool as pool
-import appscale_logger
-import pycassa
-from thrift import Thrift
-from thrift.transport import TSocket
-from thrift.transport import TTransport
-from thrift.protocol import TBinaryProtocol
-ERROR_DEFAULT = "DB_ERROR:" # ERROR_CASSANDRA
-# Store all schema information in a special table
-# If a table does not show up in this table, try a range query 
-# to discover it's schema
-SCHEMA_TABLE = "__key__"
-SCHEMA_TABLE_SCHEMA = ['schema']
-# use 1 Table and 1 ColumnFamily in Cassandra
-MAIN_TABLE = "Keyspace1"
-COLUMN_FAMILY = "Standard1"
-
-PERSISTENT_CONNECTION = False
-PROFILING = False
-
-DEFAULT_HOST = "localhost"
-DEFAULT_PORT = 9160
-
-CONSISTENCY_ZERO = 0 # don't use this for reads
-CONSISTENCY_ONE = 1
-CONSISTENCY_QUORUM = 2
-CONSISTENCY_ALL = 5 # don't use this for reads (next version may fix this)
-
-MAX_ROW_COUNT = 10000000
-table_cache = {}
-class DatastoreProxy(AppDBInterface):
-  def __init__(self, logger = appscale_logger.getLogger("datastore-cassandra")):
-    f = open(APPSCALE_HOME + '/.appscale/my_private_ip', 'r')
-    self.host = DEFAULT_HOST
-    self.port = DEFAULT_PORT
-    #self.pool = pool.QueuePool(self.__create_connection, reset_on_return=False)
-    #connection.add_pool('AppScale', [self.host, self.port])
-    self.pool = pycassa.ConnectionPool(keyspace='Keyspace1', server_list=[self.host+":"+str(self.port)], prefill=False)
-    self.logger = logger
-
-  def logTiming(self, function, start_time, end_time):
-    if PROFILING:
-      self.logger.debug(function + ": " + str(end_time - start_time) + " s")
-  
-  def get_entity(self, table_name, row_key, column_names):
-    error = [ERROR_DEFAULT]
-    list = error
-    client = None
-    row_key = table_name + '/' + row_key
-    try: 
-      slice_predicate = SlicePredicate(column_names=column_names)
-      path = ColumnPath(COLUMN_FAMILY)
-      client = self.__setup_connection()
-      # Result is a column type which has name, value, timestamp
-      result = client.get_slice(row_key, path, slice_predicate,
-                                 CONSISTENCY_QUORUM) 
-      for column in column_names:
-        for r in result:
-          c = r.column
-          if column == c.name:
-            list.append(c.value)
-    except NotFoundException: # occurs normally if the item isn't in the db 
-      list[0] += "Not found"
-      self.__close_connection(client)
-      return list
-    except Exception, ex:
-      #self.logger.debug("Exception %s" % ex)
-      list[0]+=("Exception: %s"%ex)
-      self.__close_connection(client)
-      return list
-    self.__close_connection(client)
-    if len(list) == 1:
-      list[0] += "Not found"
-    return list
-
-
-  def put_entity(self, table_name, row_key, column_names, cell_values):
-    error = [ERROR_DEFAULT]
-    list = error
-    client = None
-
-    # The first time a table is seen
-    if table_name not in table_cache:
-      self.create_table(table_name, column_names)
-
-    row_key = table_name + '/' + row_key
-    client = self.__setup_connection()
-    curtime = self.timestamp()
-    # Result is a column type which has name, value, timestamp
-    mutations = []
-    for index, ii in enumerate(column_names):
-      column = Column(name = ii, value=cell_values[index],
-                    timestamp=curtime)
-      c_or_sc = ColumnOrSuperColumn(column=column)
-      mutation = Mutation(column_or_supercolumn=c_or_sc)
-      mutations.append(mutation)
-    mutation_map = {row_key : { COLUMN_FAMILY : mutations } }
-    client.batch_mutate(mutation_map, CONSISTENCY_QUORUM) 
-    """except Exception, ex:
-      print "EXCEPTION"
-      self.logger.debug("Exception %s" % ex)
-      list[0]+=("Exception: %s"%ex)
-      self.__close_connection(client)
-      list.append("0")
-      return list
-    """
-    self.__close_connection(client)
-    list.append("0")
-    return list
-
-  def put_entity_dict(self, table_name, row_key, value_dict):
-    raise NotImplementedError("put_entity_dict is not implemented in %s." % self.__class__)
-
-
-  def get_table(self, table_name, column_names):
-    error = [ERROR_DEFAULT]  
-    client = None
-    result = error
-    keyslices = []
-    column_parent = ColumnParent(column_family="Standard1")
-    predicate = SlicePredicate(column_names=column_names)
-    start_key = table_name + "/"
-    end_key = table_name + '/~'
-    try: 
-      client = self.__setup_connection()
-      keyslices = client.get_range_slice(column_parent, 
-                              predicate, 
-                              start_key, 
-                              end_key, 
-                              MAX_ROW_COUNT, 
-                              CONSISTENCY_QUORUM)
-    except Exception, ex:
-      self.logger.debug("Exception %s" % ex)
-      result[0] += "Exception: " + str(ex)
-      self.__close_connection(client)
-      return result
-    for keyslice in keyslices:
-      ordering_dict = {}
-      for c in keyslice.columns:
-        column = c.column
-        value = column.value
-        ordering_dict[column.name] = value
-      if len(ordering_dict) == 0:
-        continue
-      for column in column_names:
-        try:
-          result.append(ordering_dict[column])
-        except:
-          result[0] += "Key error, get_table did not return the correct schema"
-    self.__close_connection(client)
-    return result
-
-  def delete_row(self, table_name, row_key):
-    error = [ERROR_DEFAULT]
-    ret = error
-    client = None
-    row_key = table_name + '/' + row_key
-    path = ColumnPath(COLUMN_FAMILY)
-    try: 
-      client = self.__setup_connection()
-      curtime = self.timestamp()
-      # Result is a column type which has name, value, timestamp
-      client.remove(row_key, path, curtime,
-                               CONSISTENCY_QUORUM) 
-    except Exception, ex:
-      self.logger.debug("Exception %s" % ex)
-      ret[0]+=("Exception: %s"%ex)
-      self.__close_connection(client)
-      return ret 
-    self.__close_connection(client)
-    ret.append("0")
-    return ret
-
-  def get_schema(self, table_name):
-    error = [ERROR_DEFAULT]
-    result = error  
-    ret = self.get_entity(SCHEMA_TABLE, 
-                          table_name, 
-                          SCHEMA_TABLE_SCHEMA)
-    if len(ret) > 1:
-      schema = ret[1]
-    else:
-      error[0] = ret[0] + "--unable to get schema"
-      return error
-    schema = schema.split(':')
-    result = result + schema
-    return result
-
-
-  def delete_table(self, table_name):
-    error = [ERROR_DEFAULT]  
-    result = error
-    keyslices = []
-    column_parent = ColumnParent(column_family="Standard1")
-    predicate = SlicePredicate(column_names=[])
-    curtime = self.timestamp()
-    path = ColumnPath(COLUMN_FAMILY)
-    start_key = table_name + "/"
-    end_key = table_name + '/~'
-    try: 
-      client = self.__setup_connection()
-      keyslices = client.get_range_slice(column_parent, 
-                              predicate, 
-                              start_key, 
-                              end_key, 
-                              MAX_ROW_COUNT, 
-                              CONSISTENCY_QUORUM)
-    except Exception, ex:
-      self.logger.debug("Exception %s" % ex)
-      result[0]+=("Exception: %s"%ex)
-      self.__close_connection(client)
-      return result
-    keys_removed = False
-    for keyslice in keyslices:
-      row_key = keyslice.key
-      client.remove( row_key, 
-                    path, 
-                    curtime,
-                    CONSISTENCY_QUORUM) 
-      keys_removed = True
-    if table_name not in table_cache and keys_removed:
-      result[0] += "Table does not exist"
-      return  result
-    if table_name in table_cache:
-      del table_cache[table_name]
- 
-    self.__close_connection(client)
-    return result
-
-  # Only stores the schema
-  def create_table(self, table_name, column_names):
-    table_cache[table_name] = 1
-    columns = ':'.join(column_names)
-    row_key = table_name
-    # Get and make sure we are not overwriting previous schemas
-    ret = self.get_entity(SCHEMA_TABLE, row_key, SCHEMA_TABLE_SCHEMA)
-    if ret[0] != ERROR_DEFAULT:
-      self.put_entity(SCHEMA_TABLE, row_key, SCHEMA_TABLE_SCHEMA, [columns])
-
-  ######################################################################
-  # private methods 
-  ######################################################################
-  def __create_connection(self):
-    socket = TSocket.TSocket(self.host, self.port)
-    transport = TTransport.TBufferedTransport(socket)
-    protocol = TBinaryProtocol.TBinaryProtocolAccelerated(transport)
-    client = Client(protocol)
-    client.transport = transport
-    client.transport.open()
-    #keyspace = KsDef(name=MAIN_TABLE, strategy_class="org.pache.cassandra.locator.SimpleStrategy", replication_factor=1, cf_defs=cfams)
-    #client.system_add_keyspace(MAIN_TABLE)
-    client.set_keyspace(MAIN_TABLE)
-    
-    return client
-
-  def __get_connection(self):
-    return connection.get_pool("AppScale")
-
-  def __setup_connection(self):
-    return self.pool.get()
-    #return connection.get_pool("AppScale")
-    #return self.pool.connect()
-
-  def __close_connection(self, client):
-    if client:
-      client.close()
-
-  def timestamp(self):
-    return int(time.time() * 1000)
+import py_cassandra
+
+py_cassandra = py_cassandra.DatastoreProxy()
+
+columns = ["a","b","c"]
+data = ["1","2","3"]
+table_name = "hello"
+key = "1"
+print "key= " + key
+print "columns= " + str(columns)
+print "data= " + str(data)
+print "table= " + table_name
+#print py_cassandra.put_entity("__hi__", key, columns, data)
+#print py_cassandra.put_entity("__hah__", key, columns, data)
+#exit(0)
+print py_cassandra.put_entity(table_name, key, columns, data)
+ret = py_cassandra.get_entity(table_name, key, columns)
+print "doing a put then get"
+print ret
+if ret[1:] != data:
+  print "ERROR doing a put then get. Data does not match"
+  print "returned: " + str(ret)
+  print "expected: " + str(data)
+  exit(1)
+else: 
+  print "Success"
+
+ret = py_cassandra.get_schema("hello")
+print ret
+print "checking schema:"
+print ret
+if ret[1:] != columns:
+  print "ERROR in recieved schema"
+  print "returned: " + str(ret)
+  print "expected: " + str(columns)
+
+ret = py_cassandra.delete_row(table_name, key)
+print "Deleting the key %s"%key
+print ret
+
+ret = py_cassandra.get_entity(table_name, key, columns)
+print "Trying to get deleted key:"
+print ret
+print "doing a put with key %s"%key
+print py_cassandra.put_entity("hello", "1", ["a","b","c"], ["1","2","3"])
+print "doing a get table" 
+print py_cassandra.get_table("hello", ["a","b","c"])
+py_cassandra.put_entity("hello", "2", ["a","b","c"], ["4","5","6"])
+print "doing get table:"
+print py_cassandra.get_table("hello", ["a","b","c"])
+py_cassandra.put_entity("hello", "3", ["a","b","c"], ["1","2","3"])
+py_cassandra.get_table("hello", ["a","b","c"])
+
+print "TRYING TO REPLACE KEY 3"
+py_cassandra.put_entity("hello", "3", ["a","b","c"], ["1","2","3"])
+print "TRYING TO REPLACE KEY 3"
+py_cassandra.get_table("hello", ["a","b","c"])
+print "TRYING TO REPLACE KEY 3"
+ret = py_cassandra.delete_row("hello", "1")
+print "TRYING TO REPLACE KEY 3"
+ret = py_cassandra.delete_row("hello", "2")
+print "TRYING TO REPLACE KEY 3"
+ret = py_cassandra.delete_row("hello", "3")
+print "TRYING TO REPLACE KEY 3"
+py_cassandra.get_table("hello", ["a","b","c"])
+print "Deleting table:"
+print py_cassandra.delete_table("hello")
+print "deleting twice:"
+print py_cassandra.delete_table("hello")
+
+table_name = u"testing_query"
+print py_cassandra.delete_table(table_name)
+column_names = [u"c1"]
+limit = 1000
+offset = 0
+key = 0
+startrow = u"000"
+endrow = u"100"
+data = u"xxx"
+for ii in range(0, 101):
+  key = str(ii)
+  key = ("0" * (3 - len(key))) + key
+  key = unicode(key)
+  print "Adding key " + key
+  print py_cassandra.put_entity(table_name, key, column_names, [data + key])
+inclusive = 1
+notJustKeys = 0
+print "SUCCESS"
+
 
=== modified file 'AppDB/hadoop/hadoop_helper.rb'
--- AppDB/hadoop/hadoop_helper.rb	2010-12-09 21:17:29 +0000
+++ AppDB/hadoop/hadoop_helper.rb	2012-02-16 06:02:21 +0000
@@ -1,5 +1,5 @@
 require 'djinn'
-HADOOP_VER = "0.20.2"
+HADOOP_VER = "0.20.2-cdh3u3"
 HADOOP_LOC = "#{APPSCALE_HOME}/AppDB/hadoop-" + HADOOP_VER 
 HDFS_PORT = 9000
 ENABLE_HADOOP_SINGLE_NODE = true
 
=== modified file 'AppDB/hadoop/patch/hadoop-hbase.patch'
--- AppDB/hadoop/patch/hadoop-hbase.patch	2010-04-08 00:25:36 +0000
+++ AppDB/hadoop/patch/hadoop-hbase.patch	2012-02-16 06:02:21 +0000
@@ -1,18 +1,20 @@
 *** src/hdfs/org/apache/hadoop/hdfs/DistributedFileSystem.java.org	2009-04-08 22:15:30.000000000 -0700
 --- src/hdfs/org/apache/hadoop/hdfs/DistributedFileSystem.java	2010-04-07 17:12:19.000000000 -0700
 ***************
-*** 20,26 ****
+*** 20,27 ****
   
   import java.io.*;
   import java.net.*;
+  import java.util.ArrayList;
 ! 
   import org.apache.hadoop.fs.permission.FsPermission;
   import org.apache.hadoop.fs.*;
   import org.apache.hadoop.conf.Configuration;
---- 20,26 ----
+--- 20,27 ----
   
   import java.io.*;
   import java.net.*;
+  import java.util.ArrayList;
 ! import org.apache.hadoop.util.*;
   import org.apache.hadoop.fs.permission.FsPermission;
   import org.apache.hadoop.fs.*;
 
=== added file 'AppDB/hadoop/templates/hadoop'
--- AppDB/hadoop/templates/hadoop	1970-01-01 00:00:00 +0000
+++ AppDB/hadoop/templates/hadoop	2012-02-16 06:02:21 +0000
@@ -0,0 +1,468 @@
+#!/usr/bin/env bash
+# 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.
+
+
+# The Hadoop command script
+#
+# Environment Variables
+#
+#   JAVA_HOME        The java implementation to use.  Overrides JAVA_HOME.
+#
+#   HADOOP_CLASSPATH Extra Java CLASSPATH entries.
+#
+#   HADOOP_USER_CLASSPATH_FIRST      When defined, the HADOOP_CLASSPATH is 
+#                                    added in the beginning of the global
+#                                    classpath. Can be defined, for example,
+#                                    by doing 
+#                                    export HADOOP_USER_CLASSPATH_FIRST=true
+#
+#   HADOOP_HEAPSIZE  The maximum amount of heap to use, in MB. 
+#                    Default is 1000.
+#
+#   HADOOP_OPTS      Extra Java runtime options.
+#   
+#   HADOOP_NAMENODE_OPTS       These options are added to HADOOP_OPTS 
+#   HADOOP_CLIENT_OPTS         when the respective command is run.
+#   HADOOP_{COMMAND}_OPTS etc  HADOOP_JT_OPTS applies to JobTracker 
+#                              for e.g.  HADOOP_CLIENT_OPTS applies to 
+#                              more than one command (fs, dfs, fsck, 
+#                              dfsadmin etc)  
+#
+#   HADOOP_CONF_DIR  Alternate conf dir. Default is ${HADOOP_HOME}/conf.
+#
+#   HADOOP_ROOT_LOGGER The root appender. Default is INFO,console
+#
+
+bin=`dirname "$0"`
+bin=`cd "$bin"; pwd`
+
+. "$bin"/hadoop-config.sh
+
+HADOOP_IDENT_STRING=${HADOOP_IDENT_STRING:-$USER}
+
+cygwin=false
+case "`uname`" in
+CYGWIN*) cygwin=true;;
+esac
+
+# if no args specified, show usage
+if [ $# = 0 ]; then
+  echo "Usage: hadoop [--config confdir] COMMAND"
+  echo "where COMMAND is one of:"
+  echo "  namenode -format     format the DFS filesystem"
+  echo "  secondarynamenode    run the DFS secondary namenode"
+  echo "  namenode             run the DFS namenode"
+  echo "  datanode             run a DFS datanode"
+  echo "  dfsadmin             run a DFS admin client"
+  echo "  mradmin              run a Map-Reduce admin client"
+  echo "  fsck                 run a DFS filesystem checking utility"
+  echo "  fs                   run a generic filesystem user client"
+  echo "  balancer             run a cluster balancing utility"
+  echo "  fetchdt              fetch a delegation token from the NameNode"
+  echo "  jobtracker           run the MapReduce job Tracker node" 
+  echo "  pipes                run a Pipes job"
+  echo "  tasktracker          run a MapReduce task Tracker node" 
+  echo "  job                  manipulate MapReduce jobs"
+  echo "  queue                get information regarding JobQueues" 
+  echo "  version              print the version"
+  echo "  jar <jar>            run a jar file"
+  echo "  distcp <srcurl> <desturl> copy file or directories recursively"
+  echo "  archive -archiveName NAME -p <parent path> <src>* <dest> create a hadoop archive"
+  echo "  oiv                  apply the offline fsimage viewer to an fsimage"
+  echo "  classpath            prints the class path needed to get the"
+  echo "  dfsgroups            get the groups which users belong to on the Name Node"
+  echo "  mrgroups             get the groups which users belong to on the Job Tracker"
+  echo "                       Hadoop jar and the required libraries"
+  echo "  daemonlog            get/set the log level for each daemon"
+  echo " or"
+  echo "  CLASSNAME            run the class named CLASSNAME"
+  echo "Most commands print help when invoked w/o parameters."
+  exit 1
+fi
+
+# get arguments
+COMMAND=$1
+shift
+
+if [ -f "${HADOOP_CONF_DIR}/hadoop-env.sh" ]; then
+  . "${HADOOP_CONF_DIR}/hadoop-env.sh"
+fi
+
+# some Java parameters
+if [ "$JAVA_HOME" != "" ]; then
+  #echo "run java in $JAVA_HOME"
+  JAVA_HOME=$JAVA_HOME
+fi
+  
+if [ "$JAVA_HOME" = "" ]; then
+  echo "Error: JAVA_HOME is not set."
+  exit 1
+fi
+
+JAVA=$JAVA_HOME/bin/java
+JAVA_HEAP_MAX=-Xmx1000m 
+
+# check envvars which might override default args
+if [ "$HADOOP_HEAPSIZE" != "" ]; then
+  #echo "run with heapsize $HADOOP_HEAPSIZE"
+  JAVA_HEAP_MAX="-Xmx""$HADOOP_HEAPSIZE""m"
+  #echo $JAVA_HEAP_MAX
+fi
+
+# CLASSPATH initially contains $HADOOP_CONF_DIR
+CLASSPATH="${HADOOP_CONF_DIR}"
+CLASSPATH=${CLASSPATH}:$JAVA_HOME/lib/tools.jar
+if [ "$HADOOP_USER_CLASSPATH_FIRST" != "" ] && [ "$HADOOP_CLASSPATH" != "" ] ; then
+  CLASSPATH=${CLASSPATH}:${HADOOP_CLASSPATH}
+fi
+
+# for developers, add Hadoop classes to CLASSPATH
+if [ -d "$HADOOP_HOME/build/classes" ]; then
+  CLASSPATH=${CLASSPATH}:$HADOOP_HOME/build/classes
+fi
+if [ -d "$HADOOP_HOME/build/webapps" ]; then
+  CLASSPATH=${CLASSPATH}:$HADOOP_HOME/build
+fi
+if [ -d "$HADOOP_HOME/build/test/classes" ]; then
+  CLASSPATH=${CLASSPATH}:$HADOOP_HOME/build/test/classes
+fi
+if [ -d "$HADOOP_HOME/build/tools" ]; then
+  CLASSPATH=${CLASSPATH}:$HADOOP_HOME/build/tools
+fi
+
+# so that filenames w/ spaces are handled correctly in loops below
+IFS=
+
+# for releases, add core hadoop jar & webapps to CLASSPATH
+if [ -d "$HADOOP_HOME/webapps" ]; then
+  CLASSPATH=${CLASSPATH}:$HADOOP_HOME
+fi
+for f in $HADOOP_HOME/hadoop-core-*.jar; do
+  CLASSPATH=${CLASSPATH}:$f;
+done
+
+# add libs to CLASSPATH
+for f in $HADOOP_HOME/lib/*.jar; do
+  CLASSPATH=${CLASSPATH}:$f;
+done
+
+if [ -d "$HADOOP_HOME/build/ivy/lib/Hadoop/common" ]; then
+for f in $HADOOP_HOME/build/ivy/lib/Hadoop/common/*.jar; do
+  CLASSPATH=${CLASSPATH}:$f;
+done
+fi
+
+for f in $HADOOP_HOME/lib/jsp-2.1/*.jar; do
+  CLASSPATH=${CLASSPATH}:$f;
+done
+
+for f in $HADOOP_HOME/hadoop-tools-*.jar; do
+  TOOL_PATH=${TOOL_PATH}:$f;
+done
+for f in $HADOOP_HOME/build/hadoop-tools-*.jar; do
+  TOOL_PATH=${TOOL_PATH}:$f;
+done
+
+# add user-specified CLASSPATH last
+if [ "$HADOOP_USER_CLASSPATH_FIRST" = "" ] && [ "$HADOOP_CLASSPATH" != "" ]; then
+  CLASSPATH=${CLASSPATH}:${HADOOP_CLASSPATH}
+fi
+
+# default log directory & file
+if [ "$HADOOP_LOG_DIR" = "" ]; then
+  HADOOP_LOG_DIR="$HADOOP_HOME/logs"
+fi
+if [ "$HADOOP_LOGFILE" = "" ]; then
+  HADOOP_LOGFILE='hadoop.log'
+fi
+
+# default policy file for service-level authorization
+if [ "$HADOOP_POLICYFILE" = "" ]; then
+  HADOOP_POLICYFILE="hadoop-policy.xml"
+fi
+
+# restore ordinary behaviour
+unset IFS
+
+# figure out which class to run
+if [ "$COMMAND" = "classpath" ] ; then
+  if $cygwin; then
+    CLASSPATH=`cygpath -p -w "$CLASSPATH"`
+  fi
+  echo $CLASSPATH
+  exit
+elif [ "$COMMAND" = "namenode" ] ; then
+  CLASS='org.apache.hadoop.hdfs.server.namenode.NameNode'
+  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_NAMENODE_OPTS"
+elif [ "$COMMAND" = "secondarynamenode" ] ; then
+  CLASS='org.apache.hadoop.hdfs.server.namenode.SecondaryNameNode'
+  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_SECONDARYNAMENODE_OPTS"
+elif [ "$COMMAND" = "datanode" ] ; then
+  CLASS='org.apache.hadoop.hdfs.server.datanode.DataNode'
+  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_DATANODE_OPTS"
+elif [ "$COMMAND" = "fs" ] ; then
+  CLASS=org.apache.hadoop.fs.FsShell
+  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_CLIENT_OPTS"
+elif [ "$COMMAND" = "dfs" ] ; then
+  CLASS=org.apache.hadoop.fs.FsShell
+  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_CLIENT_OPTS"
+elif [ "$COMMAND" = "dfsadmin" ] ; then
+  CLASS=org.apache.hadoop.hdfs.tools.DFSAdmin
+  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_CLIENT_OPTS"
+elif [ "$COMMAND" = "mradmin" ] ; then
+  CLASS=org.apache.hadoop.mapred.tools.MRAdmin
+  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_CLIENT_OPTS"
+elif [ "$COMMAND" = "fsck" ] ; then
+  CLASS=org.apache.hadoop.hdfs.tools.DFSck
+  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_CLIENT_OPTS"
+elif [ "$COMMAND" = "balancer" ] ; then
+  CLASS=org.apache.hadoop.hdfs.server.balancer.Balancer
+  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_BALANCER_OPTS"
+elif [ "$COMMAND" = "fetchdt" ] ; then
+  CLASS=org.apache.hadoop.hdfs.tools.DelegationTokenFetcher
+elif [ "$COMMAND" = "jobtracker" ] ; then
+  CLASS=org.apache.hadoop.mapred.JobTracker
+  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_JOBTRACKER_OPTS"
+elif [ "$COMMAND" = "tasktracker" ] ; then
+  CLASS=org.apache.hadoop.mapred.TaskTracker
+  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_TASKTRACKER_OPTS"
+elif [ "$COMMAND" = "job" ] ; then
+  CLASS=org.apache.hadoop.mapred.JobClient
+  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_CLIENT_OPTS"
+elif [ "$COMMAND" = "queue" ] ; then
+  CLASS=org.apache.hadoop.mapred.JobQueueClient
+  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_CLIENT_OPTS"
+elif [ "$COMMAND" = "pipes" ] ; then
+  CLASS=org.apache.hadoop.mapred.pipes.Submitter
+  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_CLIENT_OPTS"
+elif [ "$COMMAND" = "version" ] ; then
+  CLASS=org.apache.hadoop.util.VersionInfo
+  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_CLIENT_OPTS"
+elif [ "$COMMAND" = "jar" ] ; then
+  CLASS=org.apache.hadoop.util.RunJar
+  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_CLIENT_OPTS"
+elif [ "$COMMAND" = "distcp" ] ; then
+  CLASS=org.apache.hadoop.tools.DistCp
+  CLASSPATH=${CLASSPATH}:${TOOL_PATH}
+  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_CLIENT_OPTS"
+elif [ "$COMMAND" = "daemonlog" ] ; then
+  CLASS=org.apache.hadoop.log.LogLevel
+  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_CLIENT_OPTS"
+elif [ "$COMMAND" = "archive" ] ; then
+  CLASS=org.apache.hadoop.tools.HadoopArchives
+  CLASSPATH=${CLASSPATH}:${TOOL_PATH}
+  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_CLIENT_OPTS"
+elif [ "$COMMAND" = "oiv" ] ; then
+  CLASS=org.apache.hadoop.hdfs.tools.offlineImageViewer.OfflineImageViewer
+  CLASSPATH=${CLASSPATH}:${TOOL_PATH}
+  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_CLIENT_OPTS"
+elif [ "$COMMAND" = "sampler" ] ; then
+  CLASS=org.apache.hadoop.mapred.lib.InputSampler
+  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_CLIENT_OPTS"
+elif [ "$COMMAND" = "dfsgroups" ] ; then
+  CLASS=org.apache.hadoop.hdfs.tools.GetGroups
+  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_CLIENT_OPTS"
+elif [ "$COMMAND" = "mrgroups" ] ; then
+  CLASS=org.apache.hadoop.mapred.tools.GetGroups
+  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_CLIENT_OPTS"
+elif [[ "$COMMAND" = -*  ]] ; then
+  # class and package names cannot begin with a -
+  echo "Error: No command named \`$COMMAND' was found. Perhaps you meant \`hadoop ${COMMAND#-}'"
+  exit 1
+else
+  CLASS=$COMMAND
+  HADOOP_OPTS="$HADOOP_OPTS $HADOOP_CLIENT_OPTS"
+fi
+
+# cygwin path translation
+if $cygwin; then
+  CLASSPATH=`cygpath -p -w "$CLASSPATH"`
+  HADOOP_HOME=`cygpath -w "$HADOOP_HOME"`
+  HADOOP_LOG_DIR=`cygpath -w "$HADOOP_LOG_DIR"`
+  TOOL_PATH=`cygpath -p -w "$TOOL_PATH"`
+  JAVA_LIBRARY_PATH=`cygpath -p -w "$JAVA_LIBRARY_PATH"`
+fi
+
+# setup 'java.library.path' for native-hadoop code if necessary
+if [ -d "${HADOOP_HOME}/build/native" -o -d "${HADOOP_HOME}/lib/native" -o -d "${HADOOP_HOME}/sbin" ]; then
+  JAVA_PLATFORM=`CLASSPATH=${CLASSPATH} ${JAVA} -Xmx32m ${HADOOP_JAVA_PLATFORM_OPTS} org.apache.hadoop.util.PlatformName | sed -e "s/ /_/g"`
+  
+  if [ -d "$HADOOP_HOME/build/native" ]; then
+    if [ "x$JAVA_LIBRARY_PATH" != "x" ]; then
+        JAVA_LIBRARY_PATH=${JAVA_LIBRARY_PATH}:${HADOOP_HOME}/build/native/${JAVA_PLATFORM}/lib
+    else
+        JAVA_LIBRARY_PATH=${HADOOP_HOME}/build/native/${JAVA_PLATFORM}/lib
+    fi
+  fi
+  
+  if [ -d "${HADOOP_HOME}/lib/native" ]; then
+    if [ "x$JAVA_LIBRARY_PATH" != "x" ]; then
+      JAVA_LIBRARY_PATH=${JAVA_LIBRARY_PATH}:${HADOOP_HOME}/lib/native/${JAVA_PLATFORM}
+    else
+      JAVA_LIBRARY_PATH=${HADOOP_HOME}/lib/native/${JAVA_PLATFORM}
+    fi
+  fi
+
+  _JSVC_PATH=${HADOOP_HOME}/sbin/${JAVA_PLATFORM}/jsvc
+fi
+
+# cygwin path translation
+if $cygwin; then
+  JAVA_LIBRARY_PATH=`cygpath -p "$JAVA_LIBRARY_PATH"`
+fi
+
+HADOOP_OPTS="$HADOOP_OPTS -Dhadoop.log.dir=$HADOOP_LOG_DIR"
+HADOOP_OPTS="$HADOOP_OPTS -Dhadoop.log.file=$HADOOP_LOGFILE"
+HADOOP_OPTS="$HADOOP_OPTS -Dhadoop.home.dir=$HADOOP_HOME"
+HADOOP_OPTS="$HADOOP_OPTS -Dhadoop.id.str=$HADOOP_IDENT_STRING"
+HADOOP_OPTS="$HADOOP_OPTS -Dhadoop.root.logger=${HADOOP_ROOT_LOGGER:-INFO,console}"
+if [ "x$JAVA_LIBRARY_PATH" != "x" ]; then
+  HADOOP_OPTS="$HADOOP_OPTS -Djava.library.path=$JAVA_LIBRARY_PATH"
+fi  
+HADOOP_OPTS="$HADOOP_OPTS -Dhadoop.policy.file=$HADOOP_POLICYFILE"
+
+
+###########################################################################
+# DAEMON SETTINGS
+###########################################################################
+# For any command that ends in 'node', we are starting one of the daemons.
+# In this case, we do some special processing in order to automatically
+# setuid to the correct user.
+#
+# The user itself is determined as one of the following, in descending
+# precedence:
+#  HADOOP_<node>NODE_USER variable
+#  the current userid, so long as that userid is not root
+#
+# After the above is determined, it is stored into the local variable
+# _HADOOP_DAEMON_USER
+#
+# We also need to determine the "run mode". This can be one of the following:
+#
+#  "jsvc" - only supported for the datanode - we use the jsvc wrapper in
+#           the sbin/<platform> directory in order to setuid to the target
+#           user. Requires that this script is running as root.
+#  "su" -   supported only when running as root and /bin/su exists.
+#           Uses su in order to assume the identity of the daemon user.
+#  "normal" - supported only when already running as the target user.
+###########################################################################
+if [[ "$COMMAND" == *node ]] || [[ "$COMMAND" == *tracker ]]; then
+  command_uc=$(echo $COMMAND| tr a-z A-Z)
+  user_var="HADOOP_${command_uc}_USER"
+  _HADOOP_DAEMON_USER=$(eval "echo \$$user_var")
+  _HADOOP_DAEMON_USER=${_HADOOP_DAEMON_USER:-$(id -un)}
+
+  if [ -z "$_HADOOP_DAEMON_USER" ]; then
+    echo Please specify a user to run the $COMMAND by setting $user_var
+    #exit 1
+  elif  [ "$_HADOOP_DAEMON_USER" == "root" ]; then
+    echo May not run daemons as root. Please specify $user_var
+    #exit 1
+  fi
+
+  if [ "$EUID" = "0" ] ; then
+    if [ "$COMMAND" == "datanode" ] && [ -x "$_JSVC_PATH" ]; then
+      _HADOOP_RUN_MODE="jsvc"
+    elif [ -x /bin/su ]; then
+      _HADOOP_RUN_MODE="su"
+    else
+      echo "Daemon wants to run as $_HADOOP_DAEMON_USER but script is running as root"
+      echo "and su is not available."
+      #exit 1
+    fi
+  else
+    # We must be running as the user we want to run as, if we can't use jsvc or su
+    # to drop privileges
+    if [ "$_HADOOP_DAEMON_USER" != "$(whoami)" ]; then
+      echo Daemon wants to run as $_HADOOP_DAEMON_USER but not running as that user or root.
+      #exit 1
+    fi
+    _HADOOP_RUN_MODE="normal"
+  fi
+else
+  # Normal client command
+  _HADOOP_RUN_MODE="normal"
+fi
+
+###########################################################################
+# Actually run the JVM
+###########################################################################
+case "$_HADOOP_RUN_MODE" in
+  jsvc)
+    case "$COMMAND" in
+      datanode)
+        _JSVC_STARTER_CLASS=org.apache.hadoop.hdfs.server.datanode.SecureDataNodeStarter
+      ;;
+      *)
+        echo "Cannot start $COMMAND with jsvc"
+        exit 1
+      ;;
+    esac
+
+    if [ "$_HADOOP_DAEMON_DETACHED" = "true" ]; then
+      _JSVC_FLAGS="-pidfile $_HADOOP_DAEMON_PIDFILE
+                  -errfile &1
+                  -outfile $_HADOOP_DAEMON_OUT"
+    else
+      # Even though we are trying to run a non-detached datanode,
+      # jsvc will not write to stdout/stderr, so we have to pipe
+      # it and tail the logfile.
+      _JSVC_FLAGS="-nodetach
+                   -errfile &1
+                   -outfile $HADOOP_LOG_DIR/jsvc.out"
+      echo Non-detached jsvc output piping to: $HADOOP_LOG_DIR/jsvc.out
+      touch $HADOOP_LOG_DIR/jsvc.out
+      tail -f $HADOOP_LOG_DIR/jsvc.out &
+    fi
+    unset _HADOOP_DAEMON_DETACHED
+
+    exec "$_JSVC_PATH" -Dproc_$COMMAND \
+                       $_JSVC_FLAGS \
+                       -user "$_HADOOP_DAEMON_USER" \
+                       -cp "$CLASSPATH" \
+                       $JAVA_HEAP_MAX $HADOOP_OPTS \
+                       $_JSVC_STARTER_CLASS "$@"
+  ;;
+
+  normal | su)
+    # If we need to su, tack the command into a local variable
+    if [ $_HADOOP_RUN_MODE = "su" ]; then
+      _JAVA_EXEC="su $_HADOOP_DAEMON_USER -s $JAVA --"
+    else
+      _JAVA_EXEC="$JAVA"
+    fi
+
+    if [ "$_HADOOP_DAEMON_DETACHED" = "true" ]; then
+      unset _HADOOP_DAEMON_DETACHED
+      touch $_HADOOP_DAEMON_OUT
+      nohup $_JAVA_EXEC -Dproc_$COMMAND $JAVA_HEAP_MAX $HADOOP_OPTS -classpath "$CLASSPATH" $CLASS "$@" > "$_HADOOP_DAEMON_OUT" 2>&1 < /dev/null &
+      if [ "$EUID" == "0" ]; then
+        chown $_HADOOP_DAEMON_USER $_HADOOP_DAEMON_OUT
+      fi
+      echo $! > "$_HADOOP_DAEMON_PIDFILE"
+      sleep 1
+      head "$_HADOOP_DAEMON_OUT"
+    else
+      # For normal operation, just run the command
+      exec $_JAVA_EXEC -Dproc_$COMMAND $JAVA_HEAP_MAX $HADOOP_OPTS -classpath "$CLASSPATH" $CLASS "$@"
+    fi
+  ;;
+
+  *)
+    echo Bad run mode: $_HADOOP_RUN_MODE
+    exit 1
+  ;;
+esac
 
=== modified file 'AppDB/hbase/hbase-status.sh'
--- AppDB/hbase/hbase-status.sh	2010-11-18 09:50:11 +0000
+++ AppDB/hbase/hbase-status.sh	2012-02-16 06:02:21 +0000
@@ -1,6 +1,6 @@
 #!/bin/sh
 
-val=`echo "status 'summary'" | ${APPSCALE_HOME}/AppDB/hbase/hbase-0.89.20100924/bin/hbase shell | mawk '/load$/{print $1}'`
+val=`echo "status 'summary'" | ${APPSCALE_HOME}/AppDB/hbase/hbase-0.90.4-cdh3u3/bin/hbase shell | mawk '/load$/{print $1}'`
 if [ -z "$val" ]; then
     echo "0"
 else
 
=== modified file 'AppDB/hbase/hbase_helper.rb'
--- AppDB/hbase/hbase_helper.rb	2010-12-09 21:17:29 +0000
+++ AppDB/hbase/hbase_helper.rb	2012-02-16 06:02:21 +0000
@@ -3,7 +3,7 @@
 require 'helperfunctions'
 require "#{APPSCALE_HOME}/AppDB/hadoop/hadoop_helper"
 
-HBASE_LOC = "#{APPSCALE_HOME}/AppDB/hbase/hbase-0.89.20100924"
+HBASE_LOC = "#{APPSCALE_HOME}/AppDB/hbase/hbase-0.90.4-cdh3u3"
 THRIFT_PORT = 9090
 MASTER_SERVER_PORT = 60000
 ENABLE_SINGLE_NODE = true
 
=== added file 'AppDB/hbase/patch/HMaster.java'
--- AppDB/hbase/patch/HMaster.java	1970-01-01 00:00:00 +0000
+++ AppDB/hbase/patch/HMaster.java	2012-02-16 06:02:21 +0000
@@ -0,0 +1,1171 @@
+/**
+ * Copyright 2011 The Apache Software Foundation
+ *
+ * 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.master;
+
+import java.io.IOException;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.Chore;
+import org.apache.hadoop.hbase.ClusterStatus;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.HMsg;
+import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.HServerAddress;
+import org.apache.hadoop.hbase.HServerInfo;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.MasterNotRunningException;
+import org.apache.hadoop.hbase.NotAllMetaRegionsOnlineException;
+import org.apache.hadoop.hbase.Server;
+import org.apache.hadoop.hbase.TableExistsException;
+import org.apache.hadoop.hbase.TableNotDisabledException;
+import org.apache.hadoop.hbase.TableNotFoundException;
+import org.apache.hadoop.hbase.UnknownRegionException;
+import org.apache.hadoop.hbase.catalog.CatalogTracker;
+import org.apache.hadoop.hbase.catalog.MetaEditor;
+import org.apache.hadoop.hbase.catalog.MetaReader;
+import org.apache.hadoop.hbase.client.MetaScanner;
+import org.apache.hadoop.hbase.client.MetaScanner.MetaScannerVisitor;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.executor.ExecutorService;
+import org.apache.hadoop.hbase.executor.ExecutorService.ExecutorType;
+import org.apache.hadoop.hbase.ipc.HBaseRPC;
+import org.apache.hadoop.hbase.ipc.HBaseRPCProtocolVersion;
+import org.apache.hadoop.hbase.ipc.HBaseServer;
+import org.apache.hadoop.hbase.ipc.HMasterInterface;
+import org.apache.hadoop.hbase.ipc.HMasterRegionInterface;
+import org.apache.hadoop.hbase.master.LoadBalancer.RegionPlan;
+import org.apache.hadoop.hbase.master.handler.DeleteTableHandler;
+import org.apache.hadoop.hbase.master.handler.DisableTableHandler;
+import org.apache.hadoop.hbase.master.handler.EnableTableHandler;
+import org.apache.hadoop.hbase.master.handler.ModifyTableHandler;
+import org.apache.hadoop.hbase.master.handler.ServerShutdownHandler;
+import org.apache.hadoop.hbase.master.handler.TableAddFamilyHandler;
+import org.apache.hadoop.hbase.master.handler.TableDeleteFamilyHandler;
+import org.apache.hadoop.hbase.master.handler.TableModifyFamilyHandler;
+import org.apache.hadoop.hbase.master.metrics.MasterMetrics;
+import org.apache.hadoop.hbase.monitoring.MemoryBoundedLogMessageBuffer;
+import org.apache.hadoop.hbase.monitoring.MonitoredTask;
+import org.apache.hadoop.hbase.monitoring.TaskMonitor;
+import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hadoop.hbase.regionserver.wal.HLog;
+import org.apache.hadoop.hbase.replication.regionserver.Replication;
+import org.apache.hadoop.hbase.security.User;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.InfoServer;
+import org.apache.hadoop.hbase.util.Pair;
+import org.apache.hadoop.hbase.util.Sleeper;
+import org.apache.hadoop.hbase.util.Strings;
+import org.apache.hadoop.hbase.util.Threads;
+import org.apache.hadoop.hbase.util.VersionInfo;
+import org.apache.hadoop.hbase.zookeeper.ClusterStatusTracker;
+import org.apache.hadoop.hbase.zookeeper.RegionServerTracker;
+import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
+import org.apache.hadoop.io.MapWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.net.DNS;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.Watcher;
+
+/**
+ * HMaster is the "master server" for HBase. An HBase cluster has one active
+ * master.  If many masters are started, all compete.  Whichever wins goes on to
+ * run the cluster.  All others park themselves in their constructor until
+ * master or cluster shutdown or until the active master loses its lease in
+ * zookeeper.  Thereafter, all running master jostle to take over master role.
+ *
+ * <p>The Master can be asked shutdown the cluster. See {@link #shutdown()}.  In
+ * this case it will tell all regionservers to go down and then wait on them
+ * all reporting in that they are down.  This master will then shut itself down.
+ *
+ * <p>You can also shutdown just this master.  Call {@link #stopMaster()}.
+ *
+ * @see HMasterInterface
+ * @see HMasterRegionInterface
+ * @see Watcher
+ */
+public class HMaster extends Thread
+implements HMasterInterface, HMasterRegionInterface, MasterServices, Server {
+  private static final Log LOG = LogFactory.getLog(HMaster.class.getName());
+
+  // MASTER is name of the webapp and the attribute name used stuffing this
+  //instance into web context.
+  public static final String MASTER = "master";
+
+  // The configuration for the Master
+  private final Configuration conf;
+  // server for the web ui
+  private InfoServer infoServer;
+
+  // Our zk client.
+  private ZooKeeperWatcher zooKeeper;
+  // Manager and zk listener for master election
+  private ActiveMasterManager activeMasterManager;
+  // Region server tracker
+  private RegionServerTracker regionServerTracker;
+
+  // RPC server for the HMaster
+  private final HBaseServer rpcServer;
+  // Address of the HMaster
+  private final HServerAddress address;
+  // Metrics for the HMaster
+  private final MasterMetrics metrics;
+  // file system manager for the master FS operations
+  private MasterFileSystem fileSystemManager;
+
+  // server manager to deal with region server info
+  private ServerManager serverManager;
+
+  // manager of assignment nodes in zookeeper
+  AssignmentManager assignmentManager;
+  // manager of catalog regions
+  private CatalogTracker catalogTracker;
+  // Cluster status zk tracker and local setter
+  private ClusterStatusTracker clusterStatusTracker;
+  
+  // buffer for "fatal error" notices from region servers
+  // in the cluster. This is only used for assisting
+  // operations/debugging.
+  private MemoryBoundedLogMessageBuffer rsFatals;
+
+  // This flag is for stopping this Master instance.  Its set when we are
+  // stopping or aborting
+  private volatile boolean stopped = false;
+  // Set on abort -- usually failure of our zk session.
+  private volatile boolean abort = false;
+  // flag set after we become the active master (used for testing)
+  private volatile boolean isActiveMaster = false;
+  // flag set after we complete initialization once active (used for testing)
+  private volatile boolean initialized = false;
+
+  // Instance of the hbase executor service.
+  ExecutorService executorService;
+
+  private LoadBalancer balancer;
+  private Thread balancerChore;
+  // If 'true', the balancer is 'on'.  If 'false', the balancer will not run.
+  private volatile boolean balanceSwitch = true;
+
+  private Thread catalogJanitorChore;
+  private LogCleaner logCleaner;
+
+  /**
+   * Initializes the HMaster. The steps are as follows:
+   * <p>
+   * <ol>
+   * <li>Initialize HMaster RPC and address
+   * <li>Connect to ZooKeeper.
+   * </ol>
+   * <p>
+   * Remaining steps of initialization occur in {@link #run()} so that they
+   * run in their own thread rather than within the context of the constructor.
+   * @throws InterruptedException
+   */
+  public HMaster(final Configuration conf)
+  throws IOException, KeeperException, InterruptedException {
+    this.conf = conf;
+
+    /*
+     * Determine address and initialize RPC server (but do not start).
+     * The RPC server ports can be ephemeral. Create a ZKW instance.
+     */
+    HServerAddress a = new HServerAddress(getMyAddress(this.conf));
+    int numHandlers = conf.getInt("hbase.regionserver.handler.count", 10);
+    this.rpcServer = HBaseRPC.getServer(this,
+      new Class<?>[]{HMasterInterface.class, HMasterRegionInterface.class},
+      a.getBindAddress(), a.getPort(),
+      numHandlers,
+      0, // we dont use high priority handlers in master
+      false, conf,
+      0); // this is a DNC w/o high priority handlers
+    this.address = new HServerAddress(rpcServer.getListenerAddress());
+
+    // initialize server principal (if using secure Hadoop)
+    User.login(conf, "hbase.master.keytab.file",
+        "hbase.master.kerberos.principal", this.address.getHostname());
+
+    // set the thread name now we have an address
+    setName(MASTER + "-" + this.address);
+
+    Replication.decorateMasterConfiguration(this.conf);
+
+    this.rpcServer.startThreads();
+
+    // Hack! Maps DFSClient => Master for logs.  HDFS made this
+    // config param for task trackers, but we can piggyback off of it.
+    if (this.conf.get("mapred.task.id") == null) {
+      this.conf.set("mapred.task.id", "hb_m_" + this.address.toString() +
+        "_" + System.currentTimeMillis());
+    }
+
+    this.zooKeeper = new ZooKeeperWatcher(conf, MASTER + ":" +
+        address.getPort(), this);
+
+    this.metrics = new MasterMetrics(getServerName());
+  }
+
+  /**
+   * Stall startup if we are designated a backup master; i.e. we want someone
+   * else to become the master before proceeding.
+   * @param c
+   * @param amm
+   * @throws InterruptedException
+   */
+  private static void stallIfBackupMaster(final Configuration c,
+      final ActiveMasterManager amm)
+  throws InterruptedException {
+    // If we're a backup master, stall until a primary to writes his address
+    if (!c.getBoolean(HConstants.MASTER_TYPE_BACKUP,
+      HConstants.DEFAULT_MASTER_TYPE_BACKUP)) {
+      return;
+    }
+    LOG.debug("HMaster started in backup mode.  " +
+      "Stalling until master znode is written.");
+    // This will only be a minute or so while the cluster starts up,
+    // so don't worry about setting watches on the parent znode
+    while (!amm.isActiveMaster()) {
+      LOG.debug("Waiting for master address ZNode to be written " +
+        "(Also watching cluster state node)");
+      Thread.sleep(c.getInt("zookeeper.session.timeout", 180 * 1000));
+    }
+  }
+
+  /**
+   * Main processing loop for the HMaster.
+   * <ol>
+   * <li>Block until becoming active master
+   * <li>Finish initialization via {@link #finishInitialization()}
+   * <li>Enter loop until we are stopped
+   * <li>Stop services and perform cleanup once stopped
+   * </ol>
+   */
+  @Override
+  public void run() {
+    MonitoredTask startupStatus =
+      TaskMonitor.get().createStatus("Master startup");
+    startupStatus.setDescription("Master startup");
+    try {
+      /*
+       * Block on becoming the active master.
+       *
+       * We race with other masters to write our address into ZooKeeper.  If we
+       * succeed, we are the primary/active master and finish initialization.
+       *
+       * If we do not succeed, there is another active master and we should
+       * now wait until it dies to try and become the next active master.  If we
+       * do not succeed on our first attempt, this is no longer a cluster startup.
+       */
+      this.activeMasterManager = new ActiveMasterManager(zooKeeper, address,
+          this);
+      this.zooKeeper.registerListener(activeMasterManager);
+      stallIfBackupMaster(this.conf, this.activeMasterManager);
+      this.activeMasterManager.blockUntilBecomingActiveMaster(startupStatus);
+      // We are either the active master or we were asked to shutdown
+      if (!this.stopped) {
+        finishInitialization(startupStatus);
+        loop();
+      }
+    } catch (Throwable t) {
+      abort("Unhandled exception. Starting shutdown.", t);
+    } finally {
+      startupStatus.cleanup();
+      
+      stopChores();
+      // Wait for all the remaining region servers to report in IFF we were
+      // running a cluster shutdown AND we were NOT aborting.
+      if (!this.abort && this.serverManager != null &&
+          this.serverManager.isClusterShutdown()) {
+        this.serverManager.letRegionServersShutdown();
+      }
+      stopServiceThreads();
+      // Stop services started for both backup and active masters
+      if (this.activeMasterManager != null) this.activeMasterManager.stop();
+      if (this.catalogTracker != null) this.catalogTracker.stop();
+      if (this.serverManager != null) this.serverManager.stop();
+      if (this.assignmentManager != null) this.assignmentManager.stop();
+      if (this.fileSystemManager != null) this.fileSystemManager.stop();
+      this.zooKeeper.close();
+    }
+    LOG.info("HMaster main thread exiting");
+  }
+
+  private void loop() {
+    // Check if we should stop every second.
+    Sleeper sleeper = new Sleeper(1000, this);
+    while (!this.stopped) {
+      sleeper.sleep();
+    }
+  }
+
+  /**
+   * Finish initialization of HMaster after becoming the primary master.
+   *
+   * <ol>
+   * <li>Initialize master components - file system manager, server manager,
+   *     assignment manager, region server tracker, catalog tracker, etc</li>
+   * <li>Start necessary service threads - rpc server, info server,
+   *     executor services, etc</li>
+   * <li>Set cluster as UP in ZooKeeper</li>
+   * <li>Wait for RegionServers to check-in</li>
+   * <li>Split logs and perform data recovery, if necessary</li>
+   * <li>Ensure assignment of root and meta regions<li>
+   * <li>Handle either fresh cluster start or master failover</li>
+   * </ol>
+   *
+   * @throws IOException
+   * @throws InterruptedException
+   * @throws KeeperException
+   */
+  private void finishInitialization(MonitoredTask status)
+  throws IOException, InterruptedException, KeeperException {
+
+    isActiveMaster = true;
+
+    /*
+     * We are active master now... go initialize components we need to run.
+     * Note, there may be dross in zk from previous runs; it'll get addressed
+     * below after we determine if cluster startup or failover.
+     */
+
+    status.setStatus("Initializing Master file system");
+    // TODO: Do this using Dependency Injection, using PicoContainer, Guice or Spring.
+    this.fileSystemManager = new MasterFileSystem(this, metrics);
+    this.executorService = new ExecutorService(getServerName());
+    this.rsFatals = new MemoryBoundedLogMessageBuffer(
+        conf.getLong("hbase.master.buffer.for.rs.fatals", 1*1024*1024));
+
+    this.serverManager = new ServerManager(this, this, metrics);
+
+    status.setStatus("Initializing ZK system trackers");
+    this.catalogTracker = new CatalogTracker(this.zooKeeper, this.conf,
+      this, conf.getInt("hbase.master.catalog.timeout", Integer.MAX_VALUE));
+    this.catalogTracker.start();
+
+    this.assignmentManager = new AssignmentManager(this, serverManager,
+      this.catalogTracker, this.executorService);
+    this.balancer = new LoadBalancer(conf);
+    zooKeeper.registerListenerFirst(assignmentManager);
+
+    this.regionServerTracker = new RegionServerTracker(zooKeeper, this,
+      this.serverManager);
+    this.regionServerTracker.start();
+
+    // Set the cluster as up.  If new RSs, they'll be waiting on this before
+    // going ahead with their startup.
+    this.clusterStatusTracker = new ClusterStatusTracker(getZooKeeper(), this);
+    this.clusterStatusTracker.start();
+    boolean wasUp = this.clusterStatusTracker.isClusterUp();
+    if (!wasUp) this.clusterStatusTracker.setClusterUp();
+
+    LOG.info("Server active/primary master; " + this.address +
+      ", sessionid=0x" +
+      Long.toHexString(this.zooKeeper.getZooKeeper().getSessionId()) +
+      ", cluster-up flag was=" + wasUp);
+
+    // start up all service threads.
+    status.setStatus("Initializing master service threads");
+    startServiceThreads();
+
+    // Wait for region servers to report in.  Returns count of regions.
+    int regionCount = this.serverManager.waitForRegionServers(status);
+
+    // TODO: Should do this in background rather than block master startup
+    status.setStatus("Splitting logs after master startup");
+    this.fileSystemManager.
+      splitLogAfterStartup(this.serverManager.getOnlineServers());
+
+    // Make sure root and meta assigned before proceeding.
+    assignRootAndMeta(status);
+
+    // Is this fresh start with no regions assigned or are we a master joining
+    // an already-running cluster?  If regionsCount == 0, then for sure a
+    // fresh start.  TOOD: Be fancier.  If regionsCount == 2, perhaps the
+    // 2 are .META. and -ROOT- and we should fall into the fresh startup
+    // branch below.  For now, do processFailover.
+    if (regionCount == 0) {
+      LOG.info("Master startup proceeding: cluster startup");
+      this.assignmentManager.cleanoutUnassigned();
+      this.assignmentManager.assignAllUserRegions();
+    } else {
+      LOG.info("Master startup proceeding: master failover");
+      this.assignmentManager.processFailover();
+    }
+
+    // Fixing up missing daughters if any
+    status.setStatus("Fixing up missing daughters");
+    fixupDaughters(status);
+
+    // Start balancer and meta catalog janitor after meta and regions have
+    // been assigned.
+    status.setStatus("Starting balancer and catalog janitor");
+    this.balancerChore = getAndStartBalancerChore(this);
+    this.catalogJanitorChore =
+      Threads.setDaemonThreadRunning(new CatalogJanitor(this, this));
+
+    status.markComplete("Initialization successful");
+    LOG.info("Master has completed initialization");
+    initialized = true;
+  }
+
+  /**
+   * Check <code>-ROOT-</code> and <code>.META.</code> are assigned.  If not,
+   * assign them.
+   * @throws InterruptedException
+   * @throws IOException
+   * @throws KeeperException
+   * @return Count of regions we assigned.
+   */
+  int assignRootAndMeta(MonitoredTask status)
+  throws InterruptedException, IOException, KeeperException {
+    int assigned = 0;
+    long timeout = this.conf.getLong("hbase.catalog.verification.timeout", 1000);
+
+    // Work on ROOT region.  Is it in zk in transition?
+    status.setStatus("Assigning ROOT region");
+    boolean rit = this.assignmentManager.
+      processRegionInTransitionAndBlockUntilAssigned(HRegionInfo.ROOT_REGIONINFO);
+    if (!catalogTracker.verifyRootRegionLocation(timeout)) {
+      this.assignmentManager.assignRoot();
+      this.catalogTracker.waitForRoot();
+      assigned++;
+    }
+    LOG.info("-ROOT- assigned=" + assigned + ", rit=" + rit +
+      ", location=" + catalogTracker.getRootLocation());
+
+    // Work on meta region
+    status.setStatus("Assigning META region");
+    rit = this.assignmentManager.
+      processRegionInTransitionAndBlockUntilAssigned(HRegionInfo.FIRST_META_REGIONINFO);
+    if (!this.catalogTracker.verifyMetaRegionLocation(timeout)) {
+      this.assignmentManager.assignMeta();
+      this.catalogTracker.waitForMeta();
+      // Above check waits for general meta availability but this does not
+      // guarantee that the transition has completed
+      this.assignmentManager.waitForAssignment(HRegionInfo.FIRST_META_REGIONINFO);
+      assigned++;
+    }
+    LOG.info(".META. assigned=" + assigned + ", rit=" + rit +
+      ", location=" + catalogTracker.getMetaLocation());
+    status.setStatus("META and ROOT assigned.");
+    return assigned;
+  }
+
+  void fixupDaughters(final MonitoredTask status) throws IOException {
+    final Map<HRegionInfo, Result> offlineSplitParents =
+      new HashMap<HRegionInfo, Result>();
+    // This visitor collects offline split parents in the .META. table
+    MetaReader.Visitor visitor = new MetaReader.Visitor() {
+      @Override
+      public boolean visit(Result r) throws IOException {
+        if (r == null || r.isEmpty()) return true;
+        HRegionInfo info = CatalogJanitor.getHRegionInfo(r);
+        if (info == null) return true; // Keep scanning
+        if (info.isOffline() && info.isSplit()) {
+          offlineSplitParents.put(info, r);
+        }
+        // Returning true means "keep scanning"
+        return true;
+      }
+    };
+    // Run full scan of .META. catalog table passing in our custom visitor
+    MetaReader.fullScan(this.catalogTracker, visitor);
+    // Now work on our list of found parents. See if any we can clean up.
+    int fixups = 0;
+    for (Map.Entry<HRegionInfo, Result> e : offlineSplitParents.entrySet()) {
+      fixups += ServerShutdownHandler.fixupDaughters(
+          e.getValue(), assignmentManager, catalogTracker);
+    }
+    if (fixups != 0) {
+      LOG.info("Scanned the catalog and fixed up " + fixups +
+        " missing daughter region(s)");
+    }
+  }
+
+  /*
+   * @return This masters' address.
+   * @throws UnknownHostException
+   */
+  private static String getMyAddress(final Configuration c)
+  throws UnknownHostException {
+    // Find out our address up in DNS.
+    String s = DNS.getDefaultIP(c.get("hbase.master.dns.interface","default"));
+    s += ":" + c.get(HConstants.MASTER_PORT,
+        Integer.toString(HConstants.DEFAULT_MASTER_PORT));
+    return s;
+  }
+
+  /** @return HServerAddress of the master server */
+  public HServerAddress getMasterAddress() {
+    return this.address;
+  }
+
+  public long getProtocolVersion(String protocol, long clientVersion) {
+    return HBaseRPCProtocolVersion.versionID;
+  }
+
+  /** @return InfoServer object. Maybe null.*/
+  public InfoServer getInfoServer() {
+    return this.infoServer;
+  }
+
+  @Override
+  public Configuration getConfiguration() {
+    return this.conf;
+  }
+
+  @Override
+  public ServerManager getServerManager() {
+    return this.serverManager;
+  }
+
+  @Override
+  public ExecutorService getExecutorService() {
+    return this.executorService;
+  }
+
+  @Override
+  public MasterFileSystem getMasterFileSystem() {
+    return this.fileSystemManager;
+  }
+
+  /**
+   * Get the ZK wrapper object - needed by master_jsp.java
+   * @return the zookeeper wrapper
+   */
+  public ZooKeeperWatcher getZooKeeperWatcher() {
+    return this.zooKeeper;
+  }
+
+  /*
+   * Start up all services. If any of these threads gets an unhandled exception
+   * then they just die with a logged message.  This should be fine because
+   * in general, we do not expect the master to get such unhandled exceptions
+   *  as OOMEs; it should be lightly loaded. See what HRegionServer does if
+   *  need to install an unexpected exception handler.
+   */
+  private void startServiceThreads() throws IOException{
+ 
+   // Start the executor service pools
+   this.executorService.startExecutorService(ExecutorType.MASTER_OPEN_REGION,
+      conf.getInt("hbase.master.executor.openregion.threads", 5));
+   this.executorService.startExecutorService(ExecutorType.MASTER_CLOSE_REGION,
+      conf.getInt("hbase.master.executor.closeregion.threads", 5));
+   this.executorService.startExecutorService(ExecutorType.MASTER_SERVER_OPERATIONS,
+      conf.getInt("hbase.master.executor.serverops.threads", 3));
+   this.executorService.startExecutorService(ExecutorType.MASTER_META_SERVER_OPERATIONS,
+      conf.getInt("hbase.master.executor.serverops.threads", 5));
+   
+   // We depend on there being only one instance of this executor running
+   // at a time.  To do concurrency, would need fencing of enable/disable of
+   // tables.
+   this.executorService.startExecutorService(ExecutorType.MASTER_TABLE_OPERATIONS, 1);
+
+   // Start log cleaner thread
+   String n = Thread.currentThread().getName();
+   this.logCleaner =
+      new LogCleaner(conf.getInt("hbase.master.cleaner.interval", 60 * 1000),
+         this, conf, getMasterFileSystem().getFileSystem(),
+         getMasterFileSystem().getOldLogDir());
+         Threads.setDaemonThreadRunning(logCleaner, n + ".oldLogCleaner");
+
+   // Put up info server.
+   int port = this.conf.getInt("hbase.master.info.port", 60010);
+   if (port >= 0) {
+     String a = this.conf.get("hbase.master.info.bindAddress", "0.0.0.0");
+     this.infoServer = new InfoServer(MASTER, a, port, false);
+     this.infoServer.addServlet("status", "/master-status", MasterStatusServlet.class);
+     this.infoServer.addServlet("dump", "/dump", MasterDumpServlet.class);
+     this.infoServer.setAttribute(MASTER, this);
+     this.infoServer.start();
+    }
+   
+    // Start allowing requests to happen.
+    this.rpcServer.openServer();
+    if (LOG.isDebugEnabled()) {
+      LOG.debug("Started service threads");
+    }
+
+  }
+
+  private void stopServiceThreads() {
+    if (LOG.isDebugEnabled()) {
+      LOG.debug("Stopping service threads");
+    }
+    if (this.rpcServer != null) this.rpcServer.stop();
+    // Clean up and close up shop
+    if (this.logCleaner!= null) this.logCleaner.interrupt();
+    if (this.infoServer != null) {
+      LOG.info("Stopping infoServer");
+      try {
+        this.infoServer.stop();
+      } catch (Exception ex) {
+        ex.printStackTrace();
+      }
+    }
+    if (this.executorService != null) this.executorService.shutdown();
+  }
+
+  private static Thread getAndStartBalancerChore(final HMaster master) {
+    String name = master.getServerName() + "-BalancerChore";
+    int period = master.getConfiguration().getInt("hbase.balancer.period", 300000);
+    // Start up the load balancer chore
+    Chore chore = new Chore(name, period, master) {
+      @Override
+      protected void chore() {
+        master.balance();
+      }
+    };
+    return Threads.setDaemonThreadRunning(chore);
+  }
+
+  private void stopChores() {
+    if (this.balancerChore != null) {
+      this.balancerChore.interrupt();
+    }
+    if (this.catalogJanitorChore != null) {
+      this.catalogJanitorChore.interrupt();
+    }
+  }
+
+  @Override
+  public MapWritable regionServerStartup(final HServerInfo serverInfo,
+    final long serverCurrentTime)
+  throws IOException {
+    // Set the ip into the passed in serverInfo.  Its ip is more than likely
+    // not the ip that the master sees here.  See at end of this method where
+    // we pass it back to the regionserver by setting "hbase.regionserver.address"
+    // Everafter, the HSI combination 'server name' is what uniquely identifies
+    // the incoming RegionServer.
+    InetSocketAddress address = new InetSocketAddress(
+        HBaseServer.getRemoteIp().getHostName(),
+        serverInfo.getServerAddress().getPort());
+    serverInfo.setServerAddress(new HServerAddress(address));
+
+    // Register with server manager
+    this.serverManager.regionServerStartup(serverInfo, serverCurrentTime);
+    // Send back some config info
+    MapWritable mw = createConfigurationSubset();
+     mw.put(new Text("hbase.regionserver.address"),
+         serverInfo.getServerAddress());
+    return mw;
+  }
+
+  /**
+   * @return Subset of configuration to pass initializing regionservers: e.g.
+   * the filesystem to use and root directory to use.
+   */
+  protected MapWritable createConfigurationSubset() {
+    MapWritable mw = addConfig(new MapWritable(), HConstants.HBASE_DIR);
+    return addConfig(mw, "fs.default.name");
+  }
+
+  private MapWritable addConfig(final MapWritable mw, final String key) {
+    mw.put(new Text(key), new Text(this.conf.get(key)));
+    return mw;
+  }
+
+  @Override
+  public HMsg [] regionServerReport(HServerInfo serverInfo, HMsg msgs[],
+    HRegionInfo[] mostLoadedRegions)
+  throws IOException {
+    return adornRegionServerAnswer(serverInfo,
+      this.serverManager.regionServerReport(serverInfo, msgs, mostLoadedRegions));
+  }
+  
+  @Override
+  public void reportRSFatalError(HServerInfo serverInfo,
+      String errorText) {
+    String msg = "Region server " + serverInfo + " reported a fatal error:\n"
+        + errorText;
+    LOG.error(msg);
+    rsFatals.add(msg);
+  }
+
+  /**
+   * Override if you'd add messages to return to regionserver <code>hsi</code>
+   * or to send an exception.
+   * @param msgs Messages to add to
+   * @return Messages to return to
+   * @throws IOException exceptions that were injected for the region servers
+   */
+  protected HMsg [] adornRegionServerAnswer(final HServerInfo hsi,
+      final HMsg [] msgs) throws IOException {
+    return msgs;
+  }
+
+  public boolean isMasterRunning() {
+    return !isStopped();
+  }
+
+  @Override
+  public boolean balance() {
+    // If balance not true, don't run balancer.
+    if (!this.balanceSwitch) return false;
+    synchronized (this.balancer) {
+      // Only allow one balance run at at time.
+      if (this.assignmentManager.isRegionsInTransition()) {
+        LOG.debug("Not running balancer because " +
+          this.assignmentManager.getRegionsInTransition().size() +
+          " region(s) in transition: " +
+          org.apache.commons.lang.StringUtils.
+            abbreviate(this.assignmentManager.getRegionsInTransition().toString(), 256));
+        return false;
+      }
+      if (this.serverManager.areDeadServersInProgress()) {
+        LOG.debug("Not running balancer because processing dead regionserver(s): "  +
+          this.serverManager.getDeadServers());
+        return false;
+      }
+      Map<HServerInfo, List<HRegionInfo>> assignments =
+        this.assignmentManager.getAssignments();
+      // Returned Map from AM does not include mention of servers w/o assignments.
+      for (Map.Entry<String, HServerInfo> e:
+          this.serverManager.getOnlineServers().entrySet()) {
+        HServerInfo hsi = e.getValue();
+        if (!assignments.containsKey(hsi)) {
+          assignments.put(hsi, new ArrayList<HRegionInfo>());
+        }
+      }
+      List<RegionPlan> plans = this.balancer.balanceCluster(assignments);
+      if (plans != null && !plans.isEmpty()) {
+        for (RegionPlan plan: plans) {
+          LOG.info("balance " + plan);
+          this.assignmentManager.balance(plan);
+        }
+      }
+    }
+    return true;
+  }
+
+  @Override
+  public boolean balanceSwitch(final boolean b) {
+    boolean oldValue = this.balanceSwitch;
+    this.balanceSwitch = b;
+    LOG.info("Balance=" + b);
+    return oldValue;
+  }
+
+  /**
+   * Switch for the background {@link CatalogJanitor} thread.
+   * Used for testing.  The thread will continue to run.  It will just be a noop
+   * if disabled.
+   * @param b If false, the catalog janitor won't do anything.
+   */
+  public void setCatalogJanitorEnabled(final boolean b) {
+    ((CatalogJanitor)this.catalogJanitorChore).setEnabled(b);
+  }
+
+  @Override
+  public void move(final byte[] encodedRegionName, final byte[] destServerName)
+  throws UnknownRegionException {
+    Pair<HRegionInfo, HServerInfo> p =
+      this.assignmentManager.getAssignment(encodedRegionName);
+    if (p == null)
+      throw new UnknownRegionException(Bytes.toStringBinary(encodedRegionName));
+    HRegionInfo hri = p.getFirst();
+    HServerInfo dest = null;
+    if (destServerName == null || destServerName.length == 0) {
+      LOG.info("Passed destination servername is null/empty so " +
+        "choosing a server at random");
+      this.assignmentManager.clearRegionPlan(hri);
+      // Unassign will reassign it elsewhere choosing random server.
+      this.assignmentManager.unassign(hri);
+    } else {
+      dest = this.serverManager.getServerInfo(new String(destServerName));
+      RegionPlan rp = new RegionPlan(p.getFirst(), p.getSecond(), dest);
+      this.assignmentManager.balance(rp);
+    }
+  }
+
+  public void createTable(HTableDescriptor desc, byte [][] splitKeys)
+  throws IOException {
+    createTable(desc, splitKeys, false);
+  }
+
+  public void createTable(HTableDescriptor desc, byte [][] splitKeys,
+      boolean sync)
+  throws IOException {
+    if (!isMasterRunning()) {
+      throw new MasterNotRunningException();
+    }
+    HRegionInfo [] newRegions = null;
+    if(splitKeys == null || splitKeys.length == 0) {
+      newRegions = new HRegionInfo [] { new HRegionInfo(desc, null, null) };
+    } else {
+      int numRegions = splitKeys.length + 1;
+      newRegions = new HRegionInfo[numRegions];
+      byte [] startKey = null;
+      byte [] endKey = null;
+      for(int i=0;i<numRegions;i++) {
+        endKey = (i == splitKeys.length) ? null : splitKeys[i];
+        newRegions[i] = new HRegionInfo(desc, startKey, endKey);
+        startKey = endKey;
+      }
+    }
+    int timeout = conf.getInt("hbase.client.catalog.timeout", 10000);
+    // Need META availability to create a table
+    try {
+      if(catalogTracker.waitForMeta(timeout) == null) {
+        throw new NotAllMetaRegionsOnlineException();
+      }
+    } catch (InterruptedException e) {
+      LOG.warn("Interrupted waiting for meta availability", e);
+      throw new IOException(e);
+    }
+    createTable(newRegions, sync);
+  }
+
+  private synchronized void createTable(final HRegionInfo [] newRegions,
+      final boolean sync)
+  throws IOException {
+    String tableName = newRegions[0].getTableDesc().getNameAsString();
+    if(MetaReader.tableExists(catalogTracker, tableName)) {
+      throw new TableExistsException(tableName);
+    }
+    // 1. Set table enabling flag up in zk.
+    try {
+      assignmentManager.getZKTable().setEnabledTable(tableName);
+    } catch (KeeperException e) {
+      throw new IOException("Unable to ensure that the table will be" +
+          " enabled because of a ZooKeeper issue", e);
+    }
+
+    List<HRegionInfo> regionInfos = new ArrayList<HRegionInfo>();
+    final int batchSize = this.conf.getInt("hbase.master.createtable.batchsize", 100);
+    HLog hlog = null;
+    for (int regionIdx = 0; regionIdx < newRegions.length; regionIdx++) {
+      HRegionInfo newRegion = newRegions[regionIdx];
+
+      // 2. Create HRegion
+      HRegion region = HRegion.createHRegion(newRegion,
+        fileSystemManager.getRootDir(), conf, hlog);
+
+      if (hlog == null) {
+        hlog = region.getLog();
+      }
+
+      regionInfos.add(region.getRegionInfo());
+      if (regionIdx % batchSize == 0) {
+        // 3. Insert into META
+        MetaEditor.addRegionsToMeta(catalogTracker, regionInfos);
+        regionInfos.clear();
+      }
+
+      // 4. Close the new region to flush to disk.  Close log file too.
+      region.close();
+    }
+    hlog.closeAndDelete();
+    if (regionInfos.size() > 0) {
+      MetaEditor.addRegionsToMeta(catalogTracker, regionInfos);
+    }
+
+    // 5. Trigger immediate assignment of the regions in round-robin fashion
+    if (newRegions.length == 1) {
+      this.assignmentManager.assign(newRegions[0], true);
+    } else {
+      List<HServerInfo> servers = serverManager.getOnlineServersList();
+      this.assignmentManager.bulkAssignUserRegions(newRegions, servers, sync);
+    }
+
+    // 6. If sync, wait for assignment of regions
+    if (sync) {
+      LOG.debug("Waiting for " + newRegions.length + " region(s) to be assigned");
+      for (HRegionInfo regionInfo : newRegions) {
+        try {
+          this.assignmentManager.waitForAssignment(regionInfo);
+        } catch (InterruptedException e) {
+          LOG.info("Interrupted waiting for region to be assigned during " +
+              "create table call", e);
+          Thread.currentThread().interrupt();
+          return;
+        }
+      }
+    }
+  }
+
+  private static boolean isCatalogTable(final byte [] tableName) {
+    return Bytes.equals(tableName, HConstants.ROOT_TABLE_NAME) ||
+           Bytes.equals(tableName, HConstants.META_TABLE_NAME);
+  }
+
+  public void deleteTable(final byte [] tableName) throws IOException {
+    this.executorService.submit(new DeleteTableHandler(tableName, this, this));
+  }
+
+  public void addColumn(byte [] tableName, HColumnDescriptor column)
+  throws IOException {
+    new TableAddFamilyHandler(tableName, column, this, this).process();
+  }
+
+  public void modifyColumn(byte [] tableName, HColumnDescriptor descriptor)
+  throws IOException {
+    new TableModifyFamilyHandler(tableName, descriptor, this, this).process();
+  }
+
+  public void deleteColumn(final byte [] tableName, final byte [] c)
+  throws IOException {
+    new TableDeleteFamilyHandler(tableName, c, this, this).process();
+  }
+
+  public void enableTable(final byte [] tableName) throws IOException {
+    this.executorService.submit(new EnableTableHandler(this, tableName,
+      catalogTracker, assignmentManager));
+  }
+
+  public void disableTable(final byte [] tableName) throws IOException {
+    this.executorService.submit(new DisableTableHandler(this, tableName,
+      catalogTracker, assignmentManager));
+  }
+
+  /**
+   * Return the region and current deployment for the region containing
+   * the given row. If the region cannot be found, returns null. If it
+   * is found, but not currently deployed, the second element of the pair
+   * may be null.
+   */
+  Pair<HRegionInfo,HServerAddress> getTableRegionForRow(
+      final byte [] tableName, final byte [] rowKey)
+  throws IOException {
+    final AtomicReference<Pair<HRegionInfo, HServerAddress>> result =
+      new AtomicReference<Pair<HRegionInfo, HServerAddress>>(null);
+
+    MetaScannerVisitor visitor =
+      new MetaScannerVisitor() {
+        @Override
+        public boolean processRow(Result data) throws IOException {
+          if (data == null || data.size() <= 0) {
+            return true;
+          }
+          Pair<HRegionInfo, HServerAddress> pair =
+            MetaReader.metaRowToRegionPair(data);
+          if (pair == null) {
+            return false;
+          }
+          if (!Bytes.equals(pair.getFirst().getTableDesc().getName(),
+                tableName)) {
+            return false;
+          }
+          result.set(pair);
+          return true;
+        }
+    };
+
+    MetaScanner.metaScan(conf, visitor, tableName, rowKey, 1);
+    return result.get();
+  }
+
+  @Override
+  public void modifyTable(final byte[] tableName, HTableDescriptor htd)
+  throws IOException {
+    this.executorService.submit(new ModifyTableHandler(tableName, htd, this, this));
+  }
+
+  @Override
+  public void checkTableModifiable(final byte [] tableName)
+  throws IOException {
+    String tableNameStr = Bytes.toString(tableName);
+    if (isCatalogTable(tableName)) {
+      throw new IOException("Can't modify catalog tables");
+    }
+    if (!MetaReader.tableExists(getCatalogTracker(), tableNameStr)) {
+      throw new TableNotFoundException(tableNameStr);
+    }
+    if (!getAssignmentManager().getZKTable().
+        isDisabledTable(Bytes.toString(tableName))) {
+      throw new TableNotDisabledException(tableName);
+    }
+  }
+
+  public void clearFromTransition(HRegionInfo hri) {
+    if (this.assignmentManager.isRegionInTransition(hri) != null) {
+      this.assignmentManager.clearRegionFromTransition(hri);
+    }
+  }
+  /**
+   * @return cluster status
+   */
+  public ClusterStatus getClusterStatus() {
+    ClusterStatus status = new ClusterStatus();
+    status.setHBaseVersion(VersionInfo.getVersion());
+    status.setServerInfo(serverManager.getOnlineServers().values());
+    status.setDeadServers(serverManager.getDeadServers());
+    status.setRegionsInTransition(assignmentManager.getRegionsInTransition());
+    return status;
+  }
+
+  @Override
+  public void abort(final String msg, final Throwable t) {
+    if (t != null) LOG.fatal(msg, t);
+    else LOG.fatal(msg);
+    this.abort = true;
+    stop("Aborting");
+  }
+
+  @Override
+  public ZooKeeperWatcher getZooKeeper() {
+    return zooKeeper;
+  }
+
+  @Override
+  public String getServerName() {
+    return address.toString();
+  }
+
+  @Override
+  public CatalogTracker getCatalogTracker() {
+    return catalogTracker;
+  }
+
+  @Override
+  public AssignmentManager getAssignmentManager() {
+    return this.assignmentManager;
+  }
+  
+  public MemoryBoundedLogMessageBuffer getRegionServerFatalLogBuffer() {
+    return rsFatals;
+  }
+
+  @Override
+  public void shutdown() {
+    this.serverManager.shutdownCluster();
+    try {
+      this.clusterStatusTracker.setClusterDown();
+    } catch (KeeperException e) {
+      LOG.error("ZooKeeper exception trying to set cluster as down in ZK", e);
+    }
+  }
+
+  @Override
+  public void stopMaster() {
+    stop("Stopped by " + Thread.currentThread().getName());
+  }
+
+  @Override
+  public void stop(final String why) {
+    LOG.info(why);
+    this.stopped = true;
+    // If we are a backup master, we need to interrupt wait
+    synchronized (this.activeMasterManager.clusterHasActiveMaster) {
+      this.activeMasterManager.clusterHasActiveMaster.notifyAll();
+    }
+  }
+
+  @Override
+  public boolean isStopped() {
+    return this.stopped;
+  }
+
+  /**
+   * Report whether this master is currently the active master or not.
+   * If not active master, we are parked on ZK waiting to become active.
+   *
+   * This method is used for testing.
+   *
+   * @return true if active master, false if not.
+   */
+  public boolean isActiveMaster() {
+    return isActiveMaster;
+  }
+
+  /**
+   * Report whether this master has completed with its initialization and is
+   * ready.  If ready, the master is also the active master.  A standby master
+   * is never ready.
+   *
+   * This method is used for testing.
+   *
+   * @return true if master is ready to go, false if not.
+   */
+  public boolean isInitialized() {
+    return initialized;
+  }
+
+  @Override
+  public void assign(final byte [] regionName, final boolean force)
+  throws IOException {
+    Pair<HRegionInfo, HServerAddress> pair =
+      MetaReader.getRegion(this.catalogTracker, regionName);
+    if (pair == null) throw new UnknownRegionException(Bytes.toString(regionName));
+    assignRegion(pair.getFirst());
+  }
+
+  public void assignRegion(HRegionInfo hri) {
+    assignmentManager.assign(hri, true);
+  }
+
+  @Override
+  public void unassign(final byte [] regionName, final boolean force)
+  throws IOException {
+    Pair<HRegionInfo, HServerAddress> pair =
+      MetaReader.getRegion(this.catalogTracker, regionName);
+    if (pair == null) throw new UnknownRegionException(Bytes.toStringBinary(regionName));
+    HRegionInfo hri = pair.getFirst();
+    if (force) {
+      this.assignmentManager.clearRegionFromTransition(hri);
+      assignRegion(hri);
+    } else {
+      this.assignmentManager.unassign(hri, force);
+    }
+  }
+
+  /**
+   * Utility for constructing an instance of the passed HMaster class.
+   * @param masterClass
+   * @param conf
+   * @return HMaster instance.
+   */
+  public static HMaster constructMaster(Class<? extends HMaster> masterClass,
+      final Configuration conf)  {
+    try {
+      Constructor<? extends HMaster> c =
+        masterClass.getConstructor(Configuration.class);
+      return c.newInstance(conf);
+    } catch (InvocationTargetException ite) {
+      Throwable target = ite.getTargetException() != null?
+        ite.getTargetException(): ite;
+      if (target.getCause() != null) target = target.getCause();
+      throw new RuntimeException("Failed construction of Master: " +
+        masterClass.toString(), target);
+    } catch (Exception e) {
+      throw new RuntimeException("Failed construction of Master: " +
+        masterClass.toString() + ((e.getCause() != null)?
+          e.getCause().getMessage(): ""), e);
+    }
+  }
+
+
+  /**
+   * @see org.apache.hadoop.hbase.master.HMasterCommandLine
+   */
+  public static void main(String [] args) throws Exception {
+    new HMasterCommandLine(HMaster.class).doMain(args);
+  }
+}
 
=== added file 'AppDB/hbase/patch/HRegionServer.java'
--- AppDB/hbase/patch/HRegionServer.java	1970-01-01 00:00:00 +0000
+++ AppDB/hbase/patch/HRegionServer.java	2012-02-16 06:02:21 +0000
@@ -0,0 +1,2828 @@
+/**
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.regionserver;
+
+import java.io.IOException;
+import java.lang.Thread.UncaughtExceptionHandler;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.management.ManagementFactory;
+import java.lang.management.MemoryUsage;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Method;
+import java.net.BindException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.Chore;
+import org.apache.hadoop.hbase.ClockOutOfSyncException;
+import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.HConstants.OperationStatusCode;
+import org.apache.hadoop.hbase.HMsg;
+import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.HServerAddress;
+import org.apache.hadoop.hbase.HServerInfo;
+import org.apache.hadoop.hbase.HServerLoad;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.MasterAddressTracker;
+import org.apache.hadoop.hbase.NotServingRegionException;
+import org.apache.hadoop.hbase.RemoteExceptionHandler;
+import org.apache.hadoop.hbase.Server;
+import org.apache.hadoop.hbase.Stoppable;
+import org.apache.hadoop.hbase.UnknownRowLockException;
+import org.apache.hadoop.hbase.UnknownScannerException;
+import org.apache.hadoop.hbase.YouAreDeadException;
+import org.apache.hadoop.hbase.catalog.CatalogTracker;
+import org.apache.hadoop.hbase.catalog.MetaEditor;
+import org.apache.hadoop.hbase.catalog.RootLocationEditor;
+import org.apache.hadoop.hbase.client.Action;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Increment;
+import org.apache.hadoop.hbase.client.MultiAction;
+import org.apache.hadoop.hbase.client.MultiPut;
+import org.apache.hadoop.hbase.client.MultiPutResponse;
+import org.apache.hadoop.hbase.client.MultiResponse;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.Row;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.executor.ExecutorService;
+import org.apache.hadoop.hbase.executor.ExecutorService.ExecutorType;
+import org.apache.hadoop.hbase.io.hfile.LruBlockCache;
+import org.apache.hadoop.hbase.io.hfile.LruBlockCache.CacheStats;
+import org.apache.hadoop.hbase.ipc.HBaseRPC;
+import org.apache.hadoop.hbase.ipc.HBaseRPCErrorHandler;
+import org.apache.hadoop.hbase.ipc.HBaseRPCProtocolVersion;
+import org.apache.hadoop.hbase.ipc.HBaseRpcMetrics;
+import org.apache.hadoop.hbase.ipc.HBaseServer;
+import org.apache.hadoop.hbase.ipc.HMasterRegionInterface;
+import org.apache.hadoop.hbase.ipc.HRegionInterface;
+import org.apache.hadoop.hbase.ipc.ServerNotRunningException;
+import org.apache.hadoop.hbase.regionserver.Leases.LeaseStillHeldException;
+import org.apache.hadoop.hbase.regionserver.handler.CloseMetaHandler;
+import org.apache.hadoop.hbase.regionserver.handler.CloseRegionHandler;
+import org.apache.hadoop.hbase.regionserver.handler.CloseRootHandler;
+import org.apache.hadoop.hbase.regionserver.handler.OpenMetaHandler;
+import org.apache.hadoop.hbase.regionserver.handler.OpenRegionHandler;
+import org.apache.hadoop.hbase.regionserver.handler.OpenRootHandler;
+import org.apache.hadoop.hbase.regionserver.metrics.RegionServerMetrics;
+import org.apache.hadoop.hbase.regionserver.wal.HLog;
+import org.apache.hadoop.hbase.regionserver.wal.WALObserver;
+import org.apache.hadoop.hbase.replication.regionserver.Replication;
+import org.apache.hadoop.hbase.security.User;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.CompressionTest;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.hadoop.hbase.util.FSUtils;
+import org.apache.hadoop.hbase.util.InfoServer;
+import org.apache.hadoop.hbase.util.Pair;
+import org.apache.hadoop.hbase.util.Sleeper;
+import org.apache.hadoop.hbase.util.Strings;
+import org.apache.hadoop.hbase.util.Threads;
+import org.apache.hadoop.hbase.zookeeper.ClusterStatusTracker;
+import org.apache.hadoop.hbase.zookeeper.ZKUtil;
+import org.apache.hadoop.hbase.zookeeper.ZooKeeperNodeTracker;
+import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
+import org.apache.hadoop.io.MapWritable;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.ipc.RemoteException;
+import org.apache.hadoop.net.DNS;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.zookeeper.KeeperException;
+
+import com.google.common.base.Function;
+import com.google.common.collect.Lists;
+
+/**
+ * HRegionServer makes a set of HRegions available to clients. It checks in with
+ * the HMaster. There are many HRegionServers in a single HBase deployment.
+ */
+public class HRegionServer implements HRegionInterface, HBaseRPCErrorHandler,
+    Runnable, RegionServerServices, Server {
+  public static final Log LOG = LogFactory.getLog(HRegionServer.class);
+
+  // Set when a report to the master comes back with a message asking us to
+  // shutdown. Also set by call to stop when debugging or running unit tests
+  // of HRegionServer in isolation.
+  protected volatile boolean stopped = false;
+
+  // A state before we go into stopped state.  At this stage we're closing user
+  // space regions.
+  private boolean stopping = false;
+
+  // Go down hard. Used if file system becomes unavailable and also in
+  // debugging and unit tests.
+  protected volatile boolean abortRequested;
+
+  private volatile boolean killed = false;
+
+  // If false, the file system has become unavailable
+  protected volatile boolean fsOk;
+
+  protected HServerInfo serverInfo;
+  protected final Configuration conf;
+
+  protected final AtomicBoolean haveRootRegion = new AtomicBoolean(false);
+  private FileSystem fs;
+  private Path rootDir;
+  private final Random rand = new Random();
+
+  /**
+   * Map of regions currently being served by this region server. Key is the
+   * encoded region name.  All access should be synchronized.
+   */
+  protected final Map<String, HRegion> onlineRegions =
+    new ConcurrentHashMap<String, HRegion>();
+
+  protected final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+  private final LinkedBlockingQueue<HMsg> outboundMsgs = new LinkedBlockingQueue<HMsg>();
+
+  final int numRetries;
+  protected final int threadWakeFrequency;
+  private final int msgInterval;
+
+  protected final int numRegionsToReport;
+
+  private final long maxScannerResultSize;
+
+  // Remote HMaster
+  private HMasterRegionInterface hbaseMaster;
+
+  // Server to handle client requests. Default access so can be accessed by
+  // unit tests.
+  HBaseServer server;
+
+  // Leases
+  private Leases leases;
+
+  // Request counter
+  private volatile AtomicInteger requestCount = new AtomicInteger();
+
+  // Info server. Default access so can be used by unit tests. REGIONSERVER
+  // is name of the webapp and the attribute name used stuffing this instance
+  // into web context.
+  InfoServer infoServer;
+
+  /** region server process name */
+  public static final String REGIONSERVER = "regionserver";
+
+  /*
+   * Space is reserved in HRS constructor and then released when aborting to
+   * recover from an OOME. See HBASE-706. TODO: Make this percentage of the heap
+   * or a minimum.
+   */
+  private final LinkedList<byte[]> reservedSpace = new LinkedList<byte[]>();
+
+  private RegionServerMetrics metrics;
+
+  // Compactions
+  CompactSplitThread compactSplitThread;
+
+  // Cache flushing
+  MemStoreFlusher cacheFlusher;
+
+  /*
+   * Check for major compactions.
+   */
+  Chore majorCompactionChecker;
+
+  // HLog and HLog roller. log is protected rather than private to avoid
+  // eclipse warning when accessed by inner classes
+  protected volatile HLog hlog;
+  LogRoller hlogRoller;
+
+  // flag set after we're done setting up server threads (used for testing)
+  protected volatile boolean isOnline;
+
+  final Map<String, InternalScanner> scanners = new ConcurrentHashMap<String, InternalScanner>();
+
+  // zookeeper connection and watcher
+  private ZooKeeperWatcher zooKeeper;
+
+  // master address manager and watcher
+  private MasterAddressTracker masterAddressManager;
+
+  // catalog tracker
+  private CatalogTracker catalogTracker;
+
+  // Cluster Status Tracker
+  private ClusterStatusTracker clusterStatusTracker;
+
+  // Log Splitting Worker
+  private SplitLogWorker splitLogWorker;
+
+  // A sleeper that sleeps for msgInterval.
+  private final Sleeper sleeper;
+
+  private final int rpcTimeout;
+
+  // The main region server thread.
+  @SuppressWarnings("unused")
+  private Thread regionServerThread;
+
+  // Instance of the hbase executor service.
+  private ExecutorService service;
+
+  // Replication services. If no replication, this handler will be null.
+  private Replication replicationHandler;
+
+  private final Set<byte[]> regionsInTransitionInRS =
+      new ConcurrentSkipListSet<byte[]>(Bytes.BYTES_COMPARATOR);
+
+  /**
+   * Starts a HRegionServer at the default location
+   *
+   * @param conf
+   * @throws IOException
+   * @throws InterruptedException
+   */
+  public HRegionServer(Configuration conf) throws IOException, InterruptedException {
+    this.fsOk = true;
+    this.conf = conf;
+    this.isOnline = false;
+
+    // check to see if the codec list is available:
+    String [] codecs = conf.getStrings("hbase.regionserver.codecs",
+        (String[])null);
+    if (codecs != null) {
+      for (String codec : codecs) {
+        if (!CompressionTest.testCompression(codec)) {
+          throw new IOException("Compression codec " + codec +
+              " not supported, aborting RS construction");
+        }
+      }
+    }
+
+    // Config'ed params
+    this.numRetries = conf.getInt("hbase.client.retries.number", 10);
+    this.threadWakeFrequency = conf.getInt(HConstants.THREAD_WAKE_FREQUENCY,
+        10 * 1000);
+    this.msgInterval = conf.getInt("hbase.regionserver.msginterval", 3 * 1000);
+
+    sleeper = new Sleeper(this.msgInterval, this);
+
+    this.maxScannerResultSize = conf.getLong(
+        HConstants.HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE_KEY,
+        HConstants.DEFAULT_HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE);
+
+    this.numRegionsToReport = conf.getInt(
+        "hbase.regionserver.numregionstoreport", 10);
+
+    this.rpcTimeout = conf.getInt(
+        HConstants.HBASE_RPC_TIMEOUT_KEY,
+        HConstants.DEFAULT_HBASE_RPC_TIMEOUT);
+
+    this.abortRequested = false;
+    this.stopped = false;
+
+    // Server to handle client requests
+    String machineName = DNS.getDefaultIP(
+             conf.get("hbase.regionserver.dns.interface","default"));
+
+    String addressStr = machineName + ":" +
+      conf.get(HConstants.REGIONSERVER_PORT,
+        Integer.toString(HConstants.DEFAULT_REGIONSERVER_PORT));
+    HServerAddress address = new HServerAddress(addressStr);
+    this.server = HBaseRPC.getServer(this,
+        new Class<?>[]{HRegionInterface.class, HBaseRPCErrorHandler.class,
+        OnlineRegions.class},
+        address.getBindAddress(),
+      address.getPort(), conf.getInt("hbase.regionserver.handler.count", 10),
+        conf.getInt("hbase.regionserver.metahandler.count", 10),
+        false, conf, QOS_THRESHOLD);
+    this.server.setErrorHandler(this);
+    this.server.setQosFunction(new QosFunction());
+
+    // HServerInfo can be amended by master.  See below in reportForDuty.
+    this.serverInfo = new HServerInfo(new HServerAddress(new InetSocketAddress(
+        address.getBindAddress(), this.server.getListenerAddress().getPort())),
+        System.currentTimeMillis(), this.conf.getInt(
+            "hbase.regionserver.info.port", 60030), machineName);
+    if (this.serverInfo.getServerAddress() == null) {
+      throw new NullPointerException("Server address cannot be null; "
+          + "hbase-958 debugging");
+    }
+
+    // login the server principal (if using secure Hadoop)
+    User.login(conf, "hbase.regionserver.keytab.file",
+        "hbase.regionserver.kerberos.principal", serverInfo.getHostname());
+  }
+
+  private static final int NORMAL_QOS = 0;
+  private static final int QOS_THRESHOLD = 10;  // the line between low and high qos
+  private static final int HIGH_QOS = 100;
+
+  @Retention(RetentionPolicy.RUNTIME)
+  private @interface QosPriority {
+    int priority() default 0;
+  }
+
+  class QosFunction implements Function<Writable,Integer> {
+    private final Map<String, Integer> annotatedQos;
+
+    public QosFunction() {
+      Map<String, Integer> qosMap = new HashMap<String, Integer>();
+      for (Method m : HRegionServer.class.getMethods()) {
+        QosPriority p = m.getAnnotation(QosPriority.class);
+        if (p != null) {
+          qosMap.put(m.getName(), p.priority());
+        }
+      }
+
+      annotatedQos = qosMap;
+    }
+
+    public boolean isMetaRegion(byte[] regionName) {
+      HRegion region;
+      try {
+        region = getRegion(regionName);
+      } catch (NotServingRegionException ignored) {
+        return false;
+      }
+      return region.getRegionInfo().isMetaRegion();
+    }
+
+    @Override
+    public Integer apply(Writable from) {
+      if (!(from instanceof HBaseRPC.Invocation)) return NORMAL_QOS;
+
+      HBaseRPC.Invocation inv = (HBaseRPC.Invocation) from;
+      String methodName = inv.getMethodName();
+
+      Integer priorityByAnnotation = annotatedQos.get(methodName);
+      if (priorityByAnnotation != null) {
+        return priorityByAnnotation;
+      }
+
+      // scanner methods...
+      if (methodName.equals("next") || methodName.equals("close")) {
+        // translate!
+        Long scannerId;
+        try {
+          scannerId = (Long) inv.getParameters()[0];
+        } catch (ClassCastException ignored) {
+          // LOG.debug("Low priority: " + from);
+          return NORMAL_QOS; // doh.
+        }
+        String scannerIdString = Long.toString(scannerId);
+        InternalScanner scanner = scanners.get(scannerIdString);
+        if (scanner instanceof HRegion.RegionScanner) {
+          HRegion.RegionScanner rs = (HRegion.RegionScanner) scanner;
+          HRegionInfo regionName = rs.getRegionName();
+          if (regionName.isMetaRegion()) {
+            // LOG.debug("High priority scanner request: " + scannerId);
+            return HIGH_QOS;
+          }
+        }
+      } else if (inv.getParameterClasses().length == 0) {
+       // Just let it through.  This is getOnlineRegions, etc.
+      } else if (inv.getParameterClasses()[0] == byte[].class) {
+        // first arg is byte array, so assume this is a regionname:
+        if (isMetaRegion((byte[]) inv.getParameters()[0])) {
+          // LOG.debug("High priority with method: " + methodName +
+          // " and region: "
+          // + Bytes.toString((byte[]) inv.getParameters()[0]));
+          return HIGH_QOS;
+        }
+      } else if (inv.getParameterClasses()[0] == MultiAction.class) {
+        MultiAction ma = (MultiAction) inv.getParameters()[0];
+        Set<byte[]> regions = ma.getRegions();
+        // ok this sucks, but if any single of the actions touches a meta, the
+        // whole
+        // thing gets pingged high priority. This is a dangerous hack because
+        // people
+        // can get their multi action tagged high QOS by tossing a Get(.META.)
+        // AND this
+        // regionserver hosts META/-ROOT-
+        for (byte[] region : regions) {
+          if (isMetaRegion(region)) {
+            // LOG.debug("High priority multi with region: " +
+            // Bytes.toString(region));
+            return HIGH_QOS; // short circuit for the win.
+          }
+        }
+      }
+      // LOG.debug("Low priority: " + from.toString());
+      return NORMAL_QOS;
+    }
+  }
+
+  /**
+   * Creates all of the state that needs to be reconstructed in case we are
+   * doing a restart. This is shared between the constructor and restart(). Both
+   * call it.
+   *
+   * @throws IOException
+   * @throws InterruptedException
+   */
+  private void initialize() {
+    try {
+      initializeZooKeeper();
+      initializeThreads();
+      int nbBlocks = conf.getInt("hbase.regionserver.nbreservationblocks", 4);
+      for (int i = 0; i < nbBlocks; i++) {
+        reservedSpace.add(new byte[HConstants.DEFAULT_SIZE_RESERVATION_BLOCK]);
+      }
+    } catch (Throwable t) {
+      // Call stop if error or process will stick around for ever since server
+      // puts up non-daemon threads.
+      this.server.stop();
+      abort("Initialization of RS failed.  Hence aborting RS.", t);
+    }
+  }
+
+  /**
+   * Bring up connection to zk ensemble and then wait until a master for this
+   * cluster and then after that, wait until cluster 'up' flag has been set.
+   * This is the order in which master does things.
+   * Finally put up a catalog tracker.
+   * @throws IOException
+   * @throws InterruptedException
+   */
+  private void initializeZooKeeper() throws IOException, InterruptedException {
+    // Open connection to zookeeper and set primary watcher
+    zooKeeper = new ZooKeeperWatcher(conf, REGIONSERVER + ":" +
+      serverInfo.getServerAddress().getPort(), this);
+
+    // Create the master address manager, register with zk, and start it.  Then
+    // block until a master is available.  No point in starting up if no master
+    // running.
+    this.masterAddressManager = new MasterAddressTracker(this.zooKeeper, this);
+    this.masterAddressManager.start();
+    blockAndCheckIfStopped(this.masterAddressManager);
+
+    // Wait on cluster being up.  Master will set this flag up in zookeeper
+    // when ready.
+    this.clusterStatusTracker = new ClusterStatusTracker(this.zooKeeper, this);
+    this.clusterStatusTracker.start();
+    blockAndCheckIfStopped(this.clusterStatusTracker);
+
+    // Create the catalog tracker and start it;
+    this.catalogTracker = new CatalogTracker(this.zooKeeper, this.conf,
+      this, this.conf.getInt("hbase.regionserver.catalog.timeout", Integer.MAX_VALUE));
+    catalogTracker.start();
+  }
+
+  /**
+   * Utilty method to wait indefinitely on a znode availability while checking
+   * if the region server is shut down
+   * @param tracker znode tracker to use
+   * @throws IOException any IO exception, plus if the RS is stopped
+   * @throws InterruptedException
+   */
+  private void blockAndCheckIfStopped(ZooKeeperNodeTracker tracker)
+      throws IOException, InterruptedException {
+    while (tracker.blockUntilAvailable(this.msgInterval) == null) {
+      if (this.stopped) {
+        throw new IOException("Received the shutdown message while waiting.");
+      }
+    }
+  }
+
+  /**
+   * @return False if cluster shutdown in progress
+   */
+  private boolean isClusterUp() {
+    return this.clusterStatusTracker.isClusterUp();
+  }
+
+  private void initializeThreads() throws IOException {
+
+    // Cache flushing thread.
+    this.cacheFlusher = new MemStoreFlusher(conf, this);
+
+    // Compaction thread
+    this.compactSplitThread = new CompactSplitThread(this);
+
+    // Background thread to check for major compactions; needed if region
+    // has not gotten updates in a while. Make it run at a lesser frequency.
+    int multiplier = this.conf.getInt(HConstants.THREAD_WAKE_FREQUENCY
+        + ".multiplier", 1000);
+    this.majorCompactionChecker = new MajorCompactionChecker(this,
+        this.threadWakeFrequency * multiplier, this);
+
+    this.leases = new Leases((int) conf.getLong(
+        HConstants.HBASE_REGIONSERVER_LEASE_PERIOD_KEY,
+        HConstants.DEFAULT_HBASE_REGIONSERVER_LEASE_PERIOD),
+        this.threadWakeFrequency);
+  }
+
+  /**
+   * The HRegionServer sticks in this loop until closed. It repeatedly checks in
+   * with the HMaster, sending heartbeats & reports, and receiving HRegion
+   * load/unload instructions.
+   */
+  public void run() {
+
+    try {
+      // Initialize threads and wait for a master
+      initialize();
+    } catch (Throwable e) {
+      abort("Fatal exception during initialization", e);
+    }
+
+    this.regionServerThread = Thread.currentThread();
+    try {
+      while (!this.stopped) {
+        if (tryReportForDuty()) break;
+      }
+      long lastMsg = 0;
+      List<HMsg> outboundMessages = new ArrayList<HMsg>();
+      // The main run loop.
+      for (int tries = 0; !this.stopped && isHealthy();) {
+        if (!isClusterUp()) {
+          if (isOnlineRegionsEmpty()) {
+            stop("Exiting; cluster shutdown set and not carrying any regions");
+          } else if (!this.stopping) {
+            this.stopping = true;
+            closeUserRegions(this.abortRequested);
+          } else if (this.stopping && LOG.isDebugEnabled()) {
+            LOG.debug("Waiting on " + getOnlineRegionsAsPrintableString());
+          }
+        }
+        long now = System.currentTimeMillis();
+        // Drop into the send loop if msgInterval has elapsed or if something
+        // to send. If we fail talking to the master, then we'll sleep below
+        // on poll of the outboundMsgs blockingqueue.
+        if ((now - lastMsg) >= msgInterval || !outboundMessages.isEmpty()) {
+          try {
+            doMetrics();
+            tryRegionServerReport(outboundMessages);
+            lastMsg = System.currentTimeMillis();
+            // Reset tries count if we had a successful transaction.
+            tries = 0;
+            if (this.stopped) continue;
+          } catch (Exception e) { // FindBugs REC_CATCH_EXCEPTION
+            // Two special exceptions could be printed out here,
+            // PleaseHoldException and YouAreDeadException
+            if (e instanceof IOException) {
+              e = RemoteExceptionHandler.checkIOException((IOException) e);
+            }
+            if (e instanceof YouAreDeadException) {
+              // This will be caught and handled as a fatal error below
+              throw e;
+            }
+            tries++;
+            if (tries > 0 && (tries % this.numRetries) == 0) {
+              // Check filesystem every so often.
+              checkFileSystem();
+            }
+            if (this.stopped) {
+              continue;
+            }
+            LOG.warn("Attempt=" + tries, e);
+            // No point retrying immediately; this is probably connection to
+            // master issue. Doing below will cause us to sleep.
+            lastMsg = System.currentTimeMillis();
+          }
+        }
+        now = System.currentTimeMillis();
+        HMsg msg = this.outboundMsgs.poll((msgInterval - (now - lastMsg)), TimeUnit.MILLISECONDS);
+        if (msg != null) outboundMessages.add(msg);
+      } // for
+    } catch (Throwable t) {
+      if (!checkOOME(t)) {
+        abort("Unhandled exception: " + t.getMessage(), t);
+      }
+    }
+    this.leases.closeAfterLeasesExpire();
+    this.server.stop();
+    if (this.splitLogWorker != null) {
+      splitLogWorker.stop();
+    }
+    if (this.infoServer != null) {
+      LOG.info("Stopping infoServer");
+      try {
+        this.infoServer.stop();
+      } catch (Exception e) {
+        e.printStackTrace();
+      }
+    }
+    // Send cache a shutdown.
+    LruBlockCache c = (LruBlockCache) StoreFile.getBlockCache(this.conf);
+    if (c != null) {
+      c.shutdown();
+    }
+
+    // Send interrupts to wake up threads if sleeping so they notice shutdown.
+    // TODO: Should we check they are alive? If OOME could have exited already
+    if (this.cacheFlusher != null) this.cacheFlusher.interruptIfNecessary();
+    if (this.compactSplitThread != null) this.compactSplitThread.interruptIfNecessary();
+    if (this.hlogRoller != null) this.hlogRoller.interruptIfNecessary();
+    if (this.majorCompactionChecker != null) this.majorCompactionChecker.interrupt();
+
+    if (this.killed) {
+      // Just skip out w/o closing regions.
+    } else if (abortRequested) {
+      if (this.fsOk) {
+        closeAllRegions(abortRequested); // Don't leave any open file handles
+        closeWAL(false);
+      }
+      LOG.info("aborting server at: " + this.serverInfo.getServerName());
+    } else {
+      closeAllRegions(abortRequested);
+      closeWAL(true);
+      closeAllScanners();
+      LOG.info("stopping server at: " + this.serverInfo.getServerName());
+    }
+    // Interrupt catalog tracker here in case any regions being opened out in
+    // handlers are stuck waiting on meta or root.
+    if (this.catalogTracker != null) this.catalogTracker.stop();
+    if (this.fsOk)
+      waitOnAllRegionsToClose(abortRequested);
+
+    // Make sure the proxy is down.
+    if (this.hbaseMaster != null) {
+      HBaseRPC.stopProxy(this.hbaseMaster);
+      this.hbaseMaster = null;
+    }
+    this.leases.close();
+    this.zooKeeper.close();
+    if (!killed) {
+      join();
+    }
+    LOG.info(Thread.currentThread().getName() + " exiting");
+  }
+
+  String getOnlineRegionsAsPrintableString() {
+    StringBuilder sb = new StringBuilder();
+    for (HRegion r: this.onlineRegions.values()) {
+      if (sb.length() > 0) sb.append(", ");
+      sb.append(r.getRegionInfo().getEncodedName());
+    }
+    return sb.toString();
+  }
+
+  /**
+   * Wait on regions close.
+   */
+  private void waitOnAllRegionsToClose(final boolean abort) {
+    // Wait till all regions are closed before going out.
+    int lastCount = -1;
+    while (!isOnlineRegionsEmpty()) {
+      int count = getNumberOfOnlineRegions();
+      // Only print a message if the count of regions has changed.
+      if (count != lastCount) {
+        lastCount = count;
+        LOG.info("Waiting on " + count + " regions to close");
+        // Only print out regions still closing if a small number else will
+        // swamp the log.
+        if (count < 10 && LOG.isDebugEnabled()) {
+          LOG.debug(this.onlineRegions);
+        }
+      }
+      // Ensure all user regions have been sent to close. Use this to
+      // protect against the case where an open comes in after we start the
+      // iterator of onlineRegions to close all user regions.
+      for (Map.Entry<String, HRegion> e : this.onlineRegions.entrySet()) {
+        HRegionInfo hri = e.getValue().getRegionInfo();
+        if (!this.regionsInTransitionInRS.contains(hri.getEncodedNameAsBytes())) {
+          // Don't update zk with this close transition; pass false.
+          closeRegion(hri, abort, false);
+        }
+      }
+      Threads.sleep(1000);
+    }
+  }
+
+  List<HMsg> tryRegionServerReport(final List<HMsg> outboundMessages)
+  throws IOException {
+    this.serverInfo.setLoad(buildServerLoad());
+    this.requestCount.set(0);
+    addOutboundMsgs(outboundMessages);
+    HMsg [] msgs = null;
+    while (!this.stopped) {
+      try {
+        msgs = this.hbaseMaster.regionServerReport(this.serverInfo,
+          outboundMessages.toArray(HMsg.EMPTY_HMSG_ARRAY),
+          getMostLoadedRegions());
+        break;
+      } catch (IOException ioe) {
+        if (ioe instanceof RemoteException) {
+          ioe = ((RemoteException)ioe).unwrapRemoteException();
+        }
+        if (ioe instanceof YouAreDeadException) {
+          // This will be caught and handled as a fatal error in run()
+          throw ioe;
+        }
+        LOG.warn("RemoteException connecting to master", ioe);
+        // Couldn't connect to the master, get location from zk and reconnect
+        // Method blocks until new master is found or we are stopped
+        getMaster();
+      }
+    }
+    updateOutboundMsgs(outboundMessages);
+    outboundMessages.clear();
+
+    for (int i = 0; !this.stopped && msgs != null && i < msgs.length; i++) {
+      LOG.info(msgs[i].toString());
+      // Intercept stop regionserver messages
+      if (msgs[i].getType().equals(HMsg.Type.STOP_REGIONSERVER)) {
+        stop("Received " + msgs[i]);
+        continue;
+      }
+      LOG.warn("NOT PROCESSING " + msgs[i] + " -- WHY IS MASTER SENDING IT TO US?");
+    }
+    return outboundMessages;
+  }
+
+  private HServerLoad buildServerLoad() {
+    MemoryUsage memory = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage();
+    HServerLoad hsl = new HServerLoad(requestCount.get(),
+      (int)(memory.getUsed() / 1024 / 1024),
+      (int) (memory.getMax() / 1024 / 1024));
+    for (HRegion r : this.onlineRegions.values()) {
+      hsl.addRegionInfo(createRegionLoad(r));
+    }
+    return hsl;
+  }
+
+  private void closeWAL(final boolean delete) {
+    try {
+      if (this.hlog != null) {
+        if (delete) {
+          hlog.closeAndDelete();
+        } else {
+          hlog.close();
+        }
+      }
+    } catch (Throwable e) {
+      LOG.error("Close and delete failed", RemoteExceptionHandler.checkThrowable(e));
+    }
+  }
+
+  private void closeAllScanners() {
+    // Close any outstanding scanners. Means they'll get an UnknownScanner
+    // exception next time they come in.
+    for (Map.Entry<String, InternalScanner> e : this.scanners.entrySet()) {
+      try {
+        e.getValue().close();
+      } catch (IOException ioe) {
+        LOG.warn("Closing scanner " + e.getKey(), ioe);
+      }
+    }
+  }
+
+  /*
+   * Add to the passed <code>msgs</code> messages to pass to the master.
+   *
+   * @param msgs Current outboundMsgs array; we'll add messages to this List.
+   */
+  private void addOutboundMsgs(final List<HMsg> msgs) {
+    if (msgs.isEmpty()) {
+      this.outboundMsgs.drainTo(msgs);
+      return;
+    }
+    OUTER: for (HMsg m : this.outboundMsgs) {
+      for (HMsg mm : msgs) {
+        // Be careful don't add duplicates.
+        if (mm.equals(m)) {
+          continue OUTER;
+        }
+      }
+      msgs.add(m);
+    }
+  }
+
+  /*
+   * Remove from this.outboundMsgs those messsages we sent the master.
+   *
+   * @param msgs Messages we sent the master.
+   */
+  private void updateOutboundMsgs(final List<HMsg> msgs) {
+    if (msgs.isEmpty()) {
+      return;
+    }
+    for (HMsg m : this.outboundMsgs) {
+      for (HMsg mm : msgs) {
+        if (mm.equals(m)) {
+          this.outboundMsgs.remove(m);
+          break;
+        }
+      }
+    }
+  }
+
+  /*
+   * Run init. Sets up hlog and starts up all server threads.
+   *
+   * @param c Extra configuration.
+   */
+  protected void handleReportForDutyResponse(final MapWritable c) throws IOException {
+    try {
+      for (Map.Entry<Writable, Writable> e : c.entrySet()) {
+
+        String key = e.getKey().toString();
+        // Use the address the master passed us
+        if (key.equals("hbase.regionserver.address")) {
+          HServerAddress hsa = (HServerAddress) e.getValue();
+          LOG.info("Master passed us address to use. Was="
+            + this.serverInfo.getServerAddress() + ", Now=" + hsa.toString());
+          this.serverInfo.setServerAddress(hsa);
+          continue;
+        }
+        String value = e.getValue().toString();
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("Config from master: " + key + "=" + value);
+        }
+        this.conf.set(key, value);
+      }
+      // hack! Maps DFSClient => RegionServer for logs.  HDFS made this
+      // config param for task trackers, but we can piggyback off of it.
+      if (this.conf.get("mapred.task.id") == null) {
+        this.conf.set("mapred.task.id",
+            "hb_rs_" + this.serverInfo.getServerName() + "_" +
+            System.currentTimeMillis());
+      }
+
+      // Master sent us hbase.rootdir to use. Should be fully qualified
+      // path with file system specification included. Set 'fs.defaultFS'
+      // to match the filesystem on hbase.rootdir else underlying hadoop hdfs
+      // accessors will be going against wrong filesystem (unless all is set
+      // to defaults).
+      this.conf.set("fs.defaultFS", this.conf.get("hbase.rootdir"));
+      // Get fs instance used by this RS
+      this.fs = FileSystem.get(this.conf);
+      this.rootDir = new Path(this.conf.get(HConstants.HBASE_DIR));
+      this.hlog = setupWALAndReplication();
+      // Init in here rather than in constructor after thread name has been set
+      this.metrics = new RegionServerMetrics();
+      startServiceThreads();
+      LOG.info("Serving as " + this.serverInfo.getServerName() +
+        ", RPC listening on " + this.server.getListenerAddress() +
+        ", sessionid=0x" +
+        Long.toHexString(this.zooKeeper.getZooKeeper().getSessionId()));
+      isOnline = true;
+    } catch (Throwable e) {
+      this.isOnline = false;
+      stop("Failed initialization");
+      throw convertThrowableToIOE(cleanup(e, "Failed init"),
+          "Region server startup failed");
+    }
+  }
+
+  /*
+   * @param r Region to get RegionLoad for.
+   *
+   * @return RegionLoad instance.
+   *
+   * @throws IOException
+   */
+  private HServerLoad.RegionLoad createRegionLoad(final HRegion r) {
+    byte[] name = r.getRegionName();
+    int stores = 0;
+    int storefiles = 0;
+    int storefileSizeMB = 0;
+    int memstoreSizeMB = (int) (r.memstoreSize.get() / 1024 / 1024);
+    int storefileIndexSizeMB = 0;
+    synchronized (r.stores) {
+      stores += r.stores.size();
+      for (Store store : r.stores.values()) {
+        storefiles += store.getStorefilesCount();
+        storefileSizeMB += (int) (store.getStorefilesSize() / 1024 / 1024);
+        storefileIndexSizeMB += (int) (store.getStorefilesIndexSize() / 1024 / 1024);
+      }
+    }
+    return new HServerLoad.RegionLoad(name, stores, storefiles,
+        storefileSizeMB, memstoreSizeMB, storefileIndexSizeMB);
+  }
+
+  /**
+   * @param encodedRegionName
+   * @return An instance of RegionLoad.
+   * @throws IOException
+   */
+  public HServerLoad.RegionLoad createRegionLoad(final String encodedRegionName) {
+    HRegion r = null;
+    r = this.onlineRegions.get(encodedRegionName);
+    return r != null ? createRegionLoad(r) : null;
+  }
+
+  /*
+   * Cleanup after Throwable caught invoking method. Converts <code>t</code> to
+   * IOE if it isn't already.
+   *
+   * @param t Throwable
+   *
+   * @return Throwable converted to an IOE; methods can only let out IOEs.
+   */
+  private Throwable cleanup(final Throwable t) {
+    return cleanup(t, null);
+  }
+
+  /*
+   * Cleanup after Throwable caught invoking method. Converts <code>t</code> to
+   * IOE if it isn't already.
+   *
+   * @param t Throwable
+   *
+   * @param msg Message to log in error. Can be null.
+   *
+   * @return Throwable converted to an IOE; methods can only let out IOEs.
+   */
+  private Throwable cleanup(final Throwable t, final String msg) {
+    // Don't log as error if NSRE; NSRE is 'normal' operation.
+    if (t instanceof NotServingRegionException) {
+      LOG.debug("NotServingRegionException; " +  t.getMessage());
+      return t;
+    }
+    if (msg == null) {
+      LOG.error("", RemoteExceptionHandler.checkThrowable(t));
+    } else {
+      LOG.error(msg, RemoteExceptionHandler.checkThrowable(t));
+    }
+    if (!checkOOME(t)) {
+      checkFileSystem();
+    }
+    return t;
+  }
+
+  /*
+   * @param t
+   *
+   * @return Make <code>t</code> an IOE if it isn't already.
+   */
+  private IOException convertThrowableToIOE(final Throwable t) {
+    return convertThrowableToIOE(t, null);
+  }
+
+  /*
+   * @param t
+   *
+   * @param msg Message to put in new IOE if passed <code>t</code> is not an IOE
+   *
+   * @return Make <code>t</code> an IOE if it isn't already.
+   */
+  private IOException convertThrowableToIOE(final Throwable t, final String msg) {
+    return (t instanceof IOException ? (IOException) t : msg == null
+        || msg.length() == 0 ? new IOException(t) : new IOException(msg, t));
+  }
+
+  /*
+   * Check if an OOME and if so, call abort.
+   *
+   * @param e
+   *
+   * @return True if we OOME'd and are aborting.
+   */
+  public boolean checkOOME(final Throwable e) {
+    boolean stop = false;
+    if (e instanceof OutOfMemoryError
+        || (e.getCause() != null && e.getCause() instanceof OutOfMemoryError)
+        || (e.getMessage() != null && e.getMessage().contains(
+            "java.lang.OutOfMemoryError"))) {
+      abort("OutOfMemoryError, aborting", e);
+      stop = true;
+    }
+    return stop;
+  }
+
+  /**
+   * Checks to see if the file system is still accessible. If not, sets
+   * abortRequested and stopRequested
+   *
+   * @return false if file system is not available
+   */
+  protected boolean checkFileSystem() {
+    if (this.fsOk && this.fs != null) {
+      try {
+        FSUtils.checkFileSystemAvailable(this.fs);
+      } catch (IOException e) {
+        abort("File System not available", e);
+        this.fsOk = false;
+      }
+    }
+    return this.fsOk;
+  }
+
+  /*
+   * Inner class that runs on a long period checking if regions need major
+   * compaction.
+   */
+  private static class MajorCompactionChecker extends Chore {
+    private final HRegionServer instance;
+    private int majorCompactPriority;
+    private final static int DEFAULT_PRIORITY = Integer.MAX_VALUE;
+
+
+    MajorCompactionChecker(final HRegionServer h, final int sleepTime,
+        final Stoppable stopper) {
+      super("MajorCompactionChecker", sleepTime, h);
+      this.instance = h;
+      LOG.info("Runs every " + sleepTime + "ms");
+
+      /*
+       * MajorCompactPriority is configurable.
+       * If not set, the compaction will use default priority.
+       */
+      majorCompactPriority = this.instance.conf.getInt(
+        "hbase.regionserver.compactionChecker.majorCompactPriority",
+        DEFAULT_PRIORITY);
+    }
+
+    @Override
+    protected void chore() {
+      for (HRegion r : this.instance.onlineRegions.values()) {
+        try {
+          if (r != null && r.isMajorCompaction()) {
+            // Queue a compaction. Will recognize if major is needed.
+        	if(majorCompactPriority == DEFAULT_PRIORITY || 
+        	    majorCompactPriority > r.getCompactPriority()){
+              this.instance.compactSplitThread.requestCompaction(r, getName()
+                + " requests major compaction use default priority");
+        	} else {
+        	  this.instance.compactSplitThread.requestCompaction(r, getName()
+                + " requests major compaction use configured priority",
+                this.majorCompactPriority);
+        	}
+          }
+        } catch (IOException e) {
+          LOG.warn("Failed major compaction check on " + r, e);
+        }
+      }
+    }
+  }
+
+  /**
+   * Report the status of the server. A server is online once all the startup is
+   * completed (setting up filesystem, starting service threads, etc.). This
+   * method is designed mostly to be useful in tests.
+   *
+   * @return true if online, false if not.
+   */
+  public boolean isOnline() {
+    return isOnline;
+  }
+
+  /**
+   * Setup WAL log and replication if enabled.
+   * Replication setup is done in here because it wants to be hooked up to WAL.
+   * @return A WAL instance.
+   * @throws IOException
+   */
+  private HLog setupWALAndReplication() throws IOException {
+    final Path oldLogDir = new Path(rootDir, HConstants.HREGION_OLDLOGDIR_NAME);
+    Path logdir = new Path(rootDir, HLog.getHLogDirectoryName(this.serverInfo));
+    if (LOG.isDebugEnabled()) {
+      LOG.debug("logdir=" + logdir);
+    }
+    if (this.fs.exists(logdir)) {
+      throw new RegionServerRunningException("Region server already "
+          + "running at " + this.serverInfo.getServerName()
+          + " because logdir " + logdir.toString() + " exists");
+    }
+
+    // Instantiate replication manager if replication enabled.  Pass it the
+    // log directories.
+    try {
+      this.replicationHandler = Replication.isReplication(this.conf)?
+        new Replication(this, this.fs, logdir, oldLogDir): null;
+    } catch (KeeperException e) {
+      throw new IOException("Failed replication handler create", e);
+    }
+    return instantiateHLog(logdir, oldLogDir);
+  }
+
+  /**
+   * Called by {@link #setupWALAndReplication()} creating WAL instance.
+   * @param logdir
+   * @param oldLogDir
+   * @return WAL instance.
+   * @throws IOException
+   */
+  protected HLog instantiateHLog(Path logdir, Path oldLogDir) throws IOException {
+    return new HLog(this.fs, logdir, oldLogDir, this.conf,
+      getWALActionListeners(), this.serverInfo.getServerAddress().toString());
+  }
+
+  /**
+   * Called by {@link #instantiateHLog(Path, Path)} setting up WAL instance.
+   * Add any {@link WALObserver}s you want inserted before WAL startup.
+   * @return List of WALActionsListener that will be passed in to
+   * {@link HLog} on construction.
+   */
+  protected List<WALObserver> getWALActionListeners() {
+    List<WALObserver> listeners = new ArrayList<WALObserver>();
+    // Log roller.
+    this.hlogRoller = new LogRoller(this, this);
+    listeners.add(this.hlogRoller);
+    if (this.replicationHandler != null) {
+      // Replication handler is an implementation of WALActionsListener.
+      listeners.add(this.replicationHandler);
+    }
+    return listeners;
+  }
+
+  protected LogRoller getLogRoller() {
+    return hlogRoller;
+  }
+
+  /*
+   * @param interval Interval since last time metrics were called.
+   */
+  protected void doMetrics() {
+    try {
+      metrics();
+    } catch (Throwable e) {
+      LOG.warn("Failed metrics", e);
+    }
+  }
+
+  protected void metrics() {
+    int seconds = this.msgInterval / 1000;
+    if(0 == seconds){
+       seconds = 1;
+    }
+    this.metrics.regions.set(this.onlineRegions.size());
+    this.metrics.requests.set(this.requestCount.get()/seconds);
+    // Is this too expensive every three seconds getting a lock on onlineRegions
+    // and then per store carried? Can I make metrics be sloppier and avoid
+    // the synchronizations?
+    int stores = 0;
+    int storefiles = 0;
+    long memstoreSize = 0;
+    long storefileIndexSize = 0;
+    for (Map.Entry<String, HRegion> e : this.onlineRegions.entrySet()) {
+        HRegion r = e.getValue();
+        memstoreSize += r.memstoreSize.get();
+        synchronized (r.stores) {
+          stores += r.stores.size();
+          for (Map.Entry<byte[], Store> ee : r.stores.entrySet()) {
+            Store store = ee.getValue();
+            storefiles += store.getStorefilesCount();
+            storefileIndexSize += store.getStorefilesIndexSize();
+          }
+        }
+      }
+    this.metrics.stores.set(stores);
+    this.metrics.storefiles.set(storefiles);
+    this.metrics.memstoreSizeMB.set((int) (memstoreSize / (1024 * 1024)));
+    this.metrics.storefileIndexSizeMB
+        .set((int) (storefileIndexSize / (1024 * 1024)));
+    this.metrics.compactionQueueSize.set(compactSplitThread
+        .getCompactionQueueSize());
+    this.metrics.flushQueueSize.set(cacheFlusher
+        .getFlushQueueSize());
+
+    LruBlockCache lruBlockCache = (LruBlockCache) StoreFile.getBlockCache(conf);
+    if (lruBlockCache != null) {
+      this.metrics.blockCacheCount.set(lruBlockCache.size());
+      this.metrics.blockCacheFree.set(lruBlockCache.getFreeSize());
+      this.metrics.blockCacheSize.set(lruBlockCache.getCurrentSize());
+      CacheStats cacheStats = lruBlockCache.getStats();
+      this.metrics.blockCacheHitCount.set(cacheStats.getHitCount());
+      this.metrics.blockCacheMissCount.set(cacheStats.getMissCount());
+      this.metrics.blockCacheEvictedCount.set(lruBlockCache.getEvictedCount());
+      double ratio = lruBlockCache.getStats().getHitRatio();
+      int percent = (int) (ratio * 100);
+      this.metrics.blockCacheHitRatio.set(percent);
+      ratio = lruBlockCache.getStats().getHitCachingRatio();
+      percent = (int) (ratio * 100);
+      this.metrics.blockCacheHitCachingRatio.set(percent);
+    }
+  }
+
+  /**
+   * @return Region server metrics instance.
+   */
+  public RegionServerMetrics getMetrics() {
+    return this.metrics;
+  }
+
+  /*
+   * Start maintanence Threads, Server, Worker and lease checker threads.
+   * Install an UncaughtExceptionHandler that calls abort of RegionServer if we
+   * get an unhandled exception. We cannot set the handler on all threads.
+   * Server's internal Listener thread is off limits. For Server, if an OOME, it
+   * waits a while then retries. Meantime, a flush or a compaction that tries to
+   * run should trigger same critical condition and the shutdown will run. On
+   * its way out, this server will shut down Server. Leases are sort of
+   * inbetween. It has an internal thread that while it inherits from Chore, it
+   * keeps its own internal stop mechanism so needs to be stopped by this
+   * hosting server. Worker logs the exception and exits.
+   */
+  private void startServiceThreads() throws IOException {
+    String n = Thread.currentThread().getName();
+    UncaughtExceptionHandler handler = new UncaughtExceptionHandler() {
+      public void uncaughtException(Thread t, Throwable e) {
+        abort("Uncaught exception in service thread " + t.getName(), e);
+      }
+    };
+
+    // Start executor services
+    this.service = new ExecutorService(getServerName());
+    this.service.startExecutorService(ExecutorType.RS_OPEN_REGION,
+      conf.getInt("hbase.regionserver.executor.openregion.threads", 3));
+    this.service.startExecutorService(ExecutorType.RS_OPEN_ROOT,
+      conf.getInt("hbase.regionserver.executor.openroot.threads", 1));
+    this.service.startExecutorService(ExecutorType.RS_OPEN_META,
+      conf.getInt("hbase.regionserver.executor.openmeta.threads", 1));
+    this.service.startExecutorService(ExecutorType.RS_CLOSE_REGION,
+      conf.getInt("hbase.regionserver.executor.closeregion.threads", 3));
+    this.service.startExecutorService(ExecutorType.RS_CLOSE_ROOT,
+      conf.getInt("hbase.regionserver.executor.closeroot.threads", 1));
+    this.service.startExecutorService(ExecutorType.RS_CLOSE_META,
+      conf.getInt("hbase.regionserver.executor.closemeta.threads", 1));
+
+    Threads.setDaemonThreadRunning(this.hlogRoller, n + ".logRoller", handler);
+    Threads.setDaemonThreadRunning(this.cacheFlusher, n + ".cacheFlusher",
+        handler);
+    Threads.setDaemonThreadRunning(this.compactSplitThread, n + ".compactor",
+        handler);
+    Threads.setDaemonThreadRunning(this.majorCompactionChecker, n
+        + ".majorCompactionChecker", handler);
+
+    // Leases is not a Thread. Internally it runs a daemon thread. If it gets
+    // an unhandled exception, it will just exit.
+    this.leases.setName(n + ".leaseChecker");
+    this.leases.start();
+    // Put up info server.
+    int port = this.conf.getInt("hbase.regionserver.info.port", 60030);
+    // -1 is for disabling info server
+    if (port >= 0) {
+      String addr = this.conf.get("hbase.regionserver.info.bindAddress",
+          "0.0.0.0");
+      // check if auto port bind enabled
+      boolean auto = this.conf.getBoolean("hbase.regionserver.info.port.auto",
+          false);
+      while (true) {
+        try {
+          this.infoServer = new InfoServer(REGIONSERVER, addr, port, false);
+          this.infoServer.addServlet("status", "/rs-status", RSStatusServlet.class);
+          this.infoServer.addServlet("dump", "/dump", RSDumpServlet.class);
+          this.infoServer.setAttribute("regionserver", this);
+          this.infoServer.start();
+          break;
+        } catch (BindException e) {
+          if (!auto) {
+            // auto bind disabled throw BindException
+            throw e;
+          }
+          // auto bind enabled, try to use another port
+          LOG.info("Failed binding http info server to port: " + port);
+          port++;
+          // update HRS server info port.
+          this.serverInfo = new HServerInfo(this.serverInfo.getServerAddress(),
+            this.serverInfo.getStartCode(), port,
+            this.serverInfo.getHostname());
+        }
+      }
+    }
+
+    if (this.replicationHandler != null) {
+      this.replicationHandler.startReplicationServices();
+    }
+
+    // Start Server.  This service is like leases in that it internally runs
+    // a thread.
+    this.server.start();
+
+    // Create the log splitting worker and start it
+    this.splitLogWorker = new SplitLogWorker(this.zooKeeper,
+        this.getConfiguration(), this.getServerName().toString());
+    splitLogWorker.start();
+  }
+
+  /*
+   * Verify that server is healthy
+   */
+  private boolean isHealthy() {
+    if (!fsOk) {
+      // File system problem
+      return false;
+    }
+    // Verify that all threads are alive
+    if (!(leases.isAlive() && compactSplitThread.isAlive()
+        && cacheFlusher.isAlive() && hlogRoller.isAlive()
+        && this.majorCompactionChecker.isAlive())) {
+      stop("One or more threads are no longer alive -- stop");
+      return false;
+    }
+    return true;
+  }
+
+  @Override
+  public HLog getWAL() {
+    return this.hlog;
+  }
+
+  @Override
+  public CatalogTracker getCatalogTracker() {
+    return this.catalogTracker;
+  }
+
+  @Override
+  public void stop(final String msg) {
+    this.stopped = true;
+    LOG.info("STOPPED: " + msg);
+    synchronized (this) {
+      // Wakes run() if it is sleeping
+      notifyAll(); // FindBugs NN_NAKED_NOTIFY
+    }
+  }
+
+  @Override
+  public void postOpenDeployTasks(final HRegion r, final CatalogTracker ct,
+      final boolean daughter)
+  throws KeeperException, IOException {
+    // Do checks to see if we need to compact (references or too many files)
+    if (r.hasReferences() || r.hasTooManyStoreFiles()) {
+      getCompactionRequester().requestCompaction(r,
+        r.hasReferences()? "Region has references on open" :
+          "Region has too many store files");
+    }
+
+    // Add to online regions if all above was successful.
+    addToOnlineRegions(r);
+
+    // Update ZK, ROOT or META
+    if (r.getRegionInfo().isRootRegion()) {
+      RootLocationEditor.setRootLocation(getZooKeeper(),
+        getServerInfo().getServerAddress());
+    } else if (r.getRegionInfo().isMetaRegion()) {
+      MetaEditor.updateMetaLocation(ct, r.getRegionInfo(), getServerInfo());
+    } else {
+      if (daughter) {
+        // If daughter of a split, update whole row, not just location.
+        MetaEditor.addDaughter(ct, r.getRegionInfo(), getServerInfo());
+      } else {
+        MetaEditor.updateRegionLocation(ct, r.getRegionInfo(), getServerInfo());
+      }
+    }
+  }
+
+  /**
+   * Return a reference to the metrics instance used for counting RPC calls.
+   * @return Metrics instance.
+   */
+  public HBaseRpcMetrics getRpcMetrics() {
+    return server.getRpcMetrics();
+  }
+
+  /**
+   * Cause the server to exit without closing the regions it is serving, the log
+   * it is using and without notifying the master. Used unit testing and on
+   * catastrophic events such as HDFS is yanked out from under hbase or we OOME.
+   *
+   * @param reason
+   *          the reason we are aborting
+   * @param cause
+   *          the exception that caused the abort, or null
+   */
+  public void abort(String reason, Throwable cause) {
+    String msg = "ABORTING region server " + this + ": " + reason;
+    if (cause != null) {
+      LOG.fatal(msg, cause);
+    } else {
+      LOG.fatal(msg);
+    }
+    this.abortRequested = true;
+    this.reservedSpace.clear();
+    if (this.metrics != null) {
+      LOG.info("Dump of metrics: " + this.metrics);
+    }
+    // Do our best to report our abort to the master, but this may not work
+    try {
+      if (cause != null) {
+        msg += "\nCause:\n" + StringUtils.stringifyException(cause);
+      }
+      if (hbaseMaster != null) {
+        hbaseMaster.reportRSFatalError(serverInfo, msg);
+      }
+    } catch (Throwable t) {
+      LOG.warn("Unable to report fatal error to master", t);
+    }
+    stop(reason);
+  }
+
+  /**
+   * @see HRegionServer#abort(String, Throwable)
+   */
+  public void abort(String reason) {
+    abort(reason, null);
+  }
+
+  /*
+   * Simulate a kill -9 of this server. Exits w/o closing regions or cleaninup
+   * logs but it does close socket in case want to bring up server on old
+   * hostname+port immediately.
+   */
+  protected void kill() {
+    this.killed = true;
+    abort("Simulated kill");
+  }
+
+  /**
+   * Wait on all threads to finish. Presumption is that all closes and stops
+   * have already been called.
+   */
+  protected void join() {
+    Threads.shutdown(this.majorCompactionChecker);
+    Threads.shutdown(this.cacheFlusher);
+    Threads.shutdown(this.compactSplitThread);
+    Threads.shutdown(this.hlogRoller);
+    this.service.shutdown();
+    if (this.replicationHandler != null) {
+      this.replicationHandler.join();
+    }
+  }
+
+  /**
+   * Get the current master from ZooKeeper and open the RPC connection to it.
+   *
+   * Method will block until a master is available. You can break from this
+   * block by requesting the server stop.
+   *
+   * @return master address, or null if server has been stopped
+   */
+  private HServerAddress getMaster() {
+    HServerAddress masterAddress = null;
+    HMasterRegionInterface master = null;
+
+    while (!stopped && master == null) {
+