浏览代码

Performed a complete manual test by stopping and starting Hive server multiple times and one at a time to ensure that the Hue code can replace the Old active HS2 with the new Active HS2. (CDPD-10924) (#1720)

Mahesh Balakrishnan 3 年之前
父节点
当前提交
17087afbb2

+ 68 - 23
apps/beeswax/src/beeswax/server/dbms.py

@@ -63,18 +63,39 @@ else:
 
 LOG = logging.getLogger(__name__)
 
+
+RESET_HS2_QUERY_SERVER = False
 DBMS_CACHE = {}
 DBMS_CACHE_LOCK = threading.Lock()
 cache = caches[CACHES_HIVE_DISCOVERY_KEY]
 
 # Using file cache to make sure eventlet threads are uniform, this cache is persistent on startup
 # So we clear it to make sure the server resets hiveserver2 host.
-cache.clear()
+def reset_ha():
+  global RESET_HS2_QUERY_SERVER
+  cache.clear()
+  RESET_HS2_QUERY_SERVER = True
+
+
+reset_ha()
+
+
+def get_zk_hs2():
+  hiveservers = None
+  zk = KazooClient(hosts=libzookeeper_conf.ENSEMBLE.get(), read_only=True)
+  zk.start()
+  znode = HIVE_DISCOVERY_HIVESERVER2_ZNODE.get()
+  if zk.exists(znode):
+    LOG.debug("Selecting up Hive server via the following node {0}".format(znode))
+    hiveservers = zk.get_children(znode)
+  zk.stop()
+  return hiveservers
 
 
 def get(user, query_server=None, cluster=None):
   global DBMS_CACHE
   global DBMS_CACHE_LOCK
+  global RESET_HS2_QUERY_SERVER
 
   if query_server is None:
     query_server = get_query_server_config(connector=cluster)
@@ -106,7 +127,15 @@ def get(user, query_server=None, cluster=None):
             HiveServerClientCompatible(HiveServerClient(query_server, user)),
             QueryHistory.SERVER_TYPE[1][0]
         )
-
+    elif RESET_HS2_QUERY_SERVER:
+      from beeswax.server.hive_server2_lib import HiveServerClient, HiveServerClientCompatible
+      RESET_HS2_QUERY_SERVER = False
+      LOG.debug('Setting DBMS cache for the new hs2')
+      DBMS_CACHE[user.id].clear()
+      DBMS_CACHE[user.id][query_server['server_name']] = HiveServer2Dbms(
+        HiveServerClientCompatible(HiveServerClient(query_server, user)),
+        QueryHistory.SERVER_TYPE[1][0]
+      )
     return DBMS_CACHE[user.id][query_server['server_name']]
   finally:
     DBMS_CACHE_LOCK.release()
@@ -172,29 +201,45 @@ def get_query_server_config(name='beeswax', connector=None):
       activeEndpoint = cache.get("hiveserver2")
       if activeEndpoint is None:
         if HIVE_DISCOVERY_HS2.get():
-          zk = KazooClient(hosts=libzookeeper_conf.ENSEMBLE.get(), read_only=True)
-          zk.start()
-          znode = HIVE_DISCOVERY_HIVESERVER2_ZNODE.get()
-          LOG.info("Selecting up Hive server via the following node {0}".format(znode))
-          if zk.exists(znode):
-            hiveservers = zk.get_children(znode)
-            LOG.info("Available Hive Servers: {0}".format(hiveservers))
-            if not hiveservers:
-              raise PopupException(_('There is no running Hive server available'))
-            server_to_use = 0  # if CONF.HIVE_SPREAD.get() randint(0, len(hiveservers)-1) else 0
-            LOG.info("Selected Hive server {0}: {1}".format(server_to_use, hiveservers[server_to_use]))
-            cache.set(
-              "hiveserver2",
-              json.dumps({
-                "host": hiveservers[server_to_use].split(";")[0].split("=")[1].split(":")[0],
-                "port": hiveservers[server_to_use].split(";")[0].split("=")[1].split(":")[1]
-              })
-            )
-          else:
-            cache.set("hiveserver2", json.dumps({"host": HIVE_SERVER_HOST.get(), "port": HIVE_HTTP_THRIFT_PORT.get()}))
-          zk.stop()
+          hiveservers = get_zk_hs2()
+          LOG.debug("Available Hive Servers: {0}".format(hiveservers))
+          if not hiveservers:
+            raise PopupException(_('There is no running Hive server available'))
+          server_to_use = 0
+          LOG.debug("Selected Hive server {0}: {1}".format(server_to_use, hiveservers[server_to_use]))
+          cache.set(
+            "hiveserver2",
+            json.dumps({
+              "host": hiveservers[server_to_use].split(";")[0].split("=")[1].split(":")[0],
+              "port": hiveservers[server_to_use].split(";")[0].split("=")[1].split(":")[1]
+            })
+          )
         else:
           cache.set("hiveserver2", json.dumps({"host": HIVE_SERVER_HOST.get(), "port": HIVE_HTTP_THRIFT_PORT.get()}))
+      else:
+        # Setting hs2 cache in-case there is no HS2 discovery
+        cache.set("hiveserver2", json.dumps({"host": HIVE_SERVER_HOST.get(), "port": HIVE_HTTP_THRIFT_PORT.get()}))
+        if HIVE_DISCOVERY_HS2.get():
+          # Replace ActiveEndpoint if the current HS2 is down
+          hiveservers = get_zk_hs2()
+          if hiveservers is not None:
+            server_to_use = 0
+            hs2_host_name = hiveservers[server_to_use].split(";")[0].split("=")[1].split(":")[0]
+            hs2_in_active_endpoint = hs2_host_name in activeEndpoint
+            LOG.debug("Is the current HS2 active {0}".format(hs2_in_active_endpoint))
+            if not hs2_in_active_endpoint:
+              LOG.error(
+                'Current HiveServer is down, working to connect with the next available HiveServer from Zookeeper')
+              reset_ha()
+              server_to_use = 0
+              LOG.debug("Selected HiveServer {0}: {1}".format(server_to_use, hiveservers[server_to_use]))
+              cache.set(
+                "hiveserver2",
+                json.dumps({
+                  "host": hiveservers[server_to_use].split(";")[0].split("=")[1].split(":")[0],
+                  "port": hiveservers[server_to_use].split(";")[0].split("=")[1].split(":")[1]
+                })
+              )
 
       activeEndpoint = json.loads(cache.get("hiveserver2"))
 

+ 44 - 26
apps/beeswax/src/beeswax/server/dbms_tests.py

@@ -18,14 +18,11 @@
 
 import logging
 import sys
-
-from django.core.cache import caches
-from nose.tools import assert_equal, assert_true, assert_raises
-
+from beeswax.server.dbms import get_query_server_config
 from desktop.lib.exceptions_renderable import PopupException
 from desktop.settings import CACHES_HIVE_DISCOVERY_KEY
-
-from beeswax.server.dbms import get_query_server_config
+from django.core.cache import caches
+from nose.tools import assert_equal, assert_raises
 
 if sys.version_info[0] > 2:
   from unittest.mock import patch, Mock
@@ -37,17 +34,15 @@ cache = caches[CACHES_HIVE_DISCOVERY_KEY]
 
 
 class TestGetQueryServerConfig():
-
   def setUp(self):
     cache.clear()
 
-
   def test_get_default(self):
 
     with patch('beeswax.conf.HIVE_SERVER_HOST.get') as HIVE_SERVER_HOST:
       with patch('beeswax.conf.HIVE_SERVER_PORT.get') as HIVE_SERVER_PORT:
         HIVE_SERVER_HOST.return_value = 'hive.gethue.com'
-        HIVE_SERVER_PORT.return_value=10002
+        HIVE_SERVER_PORT.return_value = 10002
 
         query_server = get_query_server_config()
 
@@ -55,13 +50,12 @@ class TestGetQueryServerConfig():
         assert_equal(query_server['server_host'], 'hive.gethue.com')
         assert_equal(query_server['server_port'], 10002)
 
-
   def test_get_impala(self):
 
     with patch('impala.conf.SERVER_HOST.get') as SERVER_HOST:
       with patch('impala.conf.SERVER_PORT.get') as SERVER_PORT:
         SERVER_HOST.return_value = 'impala.gethue.com'
-        SERVER_PORT.return_value=10002
+        SERVER_PORT.return_value = 10002
 
         query_server = get_query_server_config(name='impala')
 
@@ -69,13 +63,12 @@ class TestGetQueryServerConfig():
         assert_equal(query_server['server_host'], 'impala.gethue.com')
         assert_equal(query_server['server_port'], 10002)
 
-
   def test_get_llap(self):
 
     with patch('beeswax.conf.LLAP_SERVER_HOST.get') as LLAP_SERVER_HOST:
       with patch('beeswax.conf.LLAP_SERVER_PORT.get') as LLAP_SERVER_PORT:
         LLAP_SERVER_HOST.return_value = 'hive-llap.gethue.com'
-        LLAP_SERVER_PORT.return_value=10002
+        LLAP_SERVER_PORT.return_value = 10002
 
         query_server = get_query_server_config(name='llap')
 
@@ -83,18 +76,20 @@ class TestGetQueryServerConfig():
         assert_equal(query_server['server_host'], 'hive-llap.gethue.com')
         assert_equal(query_server['server_port'], 10002)
 
-
   def test_get_llap_discovery(self):
 
     with patch('beeswax.conf.HIVE_DISCOVERY_LLAP.get') as HIVE_DISCOVERY_LLAP:
       with patch('beeswax.conf.HIVE_DISCOVERY_LLAP_HA.get') as HIVE_DISCOVERY_LLAP_HA:
         with patch('beeswax.server.dbms.KazooClient') as KazooClient:
-          with patch('beeswax.conf.LLAP_SERVER_PORT.get') as LLAP_SERVER_PORT: # Workaround, to remove when assert server_port ok
+          with patch(
+              'beeswax.conf.LLAP_SERVER_PORT.get') as LLAP_SERVER_PORT:  # Workaround, to remove when assert
+            # server_port ok
             HIVE_DISCOVERY_LLAP.return_value = True
             HIVE_DISCOVERY_LLAP_HA.return_value = False
             LLAP_SERVER_PORT.return_value = 25000
             KazooClient.return_value = Mock(
-              exists=Mock(return_value=True), # Bug "TypeError: expected string or buffer" if False, to add a new test case and fix
+              exists=Mock(return_value=True),
+              # Bug "TypeError: expected string or buffer" if False, to add a new test case and fix
               get_children=Mock(return_value=['llap1=hive-llap-1.gethue.com:20000;llap2=hive-llap-2.gethue.com:20000'])
             )
             query_server = get_query_server_config(name='llap')
@@ -102,8 +97,7 @@ class TestGetQueryServerConfig():
             assert_equal(query_server['server_name'], 'beeswax')
             assert_equal(query_server['server_host'], 'hive-llap-1.gethue.com')
             # assert_equal(query_server['server_port'], 20000) # Bug Always set to LLAP_SERVER_PORT?
-            assert_equal(query_server['server_port'], 25000) # To remove this line and comment above when fixed.
-
+            assert_equal(query_server['server_port'], 25000)  # To remove this line and comment above when fixed.
 
   def test_get_llap_ha_discovery_all_server_down(self):
 
@@ -114,7 +108,8 @@ class TestGetQueryServerConfig():
           HIVE_DISCOVERY_LLAP_HA.return_value = True
 
           KazooClient.return_value = Mock(
-            exists=Mock(return_value=True), # Bug "TypeError: expected string or buffer" if False, to add a new test case and fix
+            exists=Mock(return_value=True),
+            # Bug "TypeError: expected string or buffer" if False, to add a new test case and fix
             get_children=Mock(return_value=[])
           )
 
@@ -124,7 +119,6 @@ class TestGetQueryServerConfig():
           except PopupException as e:
             assert_equal(e.message, 'There is no running Hive LLAP server available')
 
-
   def test_get_hive_ha_discovery_all_server_down(self):
 
     with patch('beeswax.conf.HIVE_DISCOVERY_LLAP.get') as HIVE_DISCOVERY_LLAP:
@@ -134,11 +128,12 @@ class TestGetQueryServerConfig():
             with patch('beeswax.server.dbms.KazooClient') as KazooClient:
               HIVE_DISCOVERY_LLAP.return_value = False
               HIVE_DISCOVERY_LLAP_HA.return_value = False
-              HIVE_DISCOVERY_HS2 = True
-              HIVE_DISCOVERY_HIVESERVER2_ZNODE = True
+              HIVE_DISCOVERY_HS2.return_value = True
+              HIVE_DISCOVERY_HIVESERVER2_ZNODE.return_value = True
 
               KazooClient.return_value = Mock(
-                exists=Mock(return_value=True), # Bug "TypeError: expected string or buffer" if False, to add a new test case and fix
+                exists=Mock(return_value=True),
+                # Bug "TypeError: expected string or buffer" if False, to add a new test case and fix
                 get_children=Mock(return_value=[])
               )
 
@@ -148,7 +143,30 @@ class TestGetQueryServerConfig():
               except PopupException as e:
                 assert_equal(e.message, 'There is no running Hive server available')
 
+  def test_get_hs2_discovery(self):
+
+    with patch('beeswax.conf.HIVE_DISCOVERY_HS2.get') as HIVE_DISCOVERY_HS2:
+      with patch('beeswax.conf.HIVE_DISCOVERY_HIVESERVER2_ZNODE.get') as HIVE_DISCOVERY_HIVESERVER2_ZNODE:
+        with patch('beeswax.server.dbms.KazooClient') as KazooClient:
+          HIVE_DISCOVERY_HS2.return_value = True
+          HIVE_DISCOVERY_HIVESERVER2_ZNODE.return_value = True
+          KazooClient.return_value = Mock(
+            exists=Mock(return_value=True),
+            # Bug "TypeError: expected string or buffer" if False, to add a new test case and fix
+            get_children=Mock(return_value=[
+              'serverUri=hive-llap-1.gethue.com:10000;serverUri=hive-llap-2.gethue.com:10000'])
+          )
+
+          try:
+            query_server = get_query_server_config(name='hive')
+          except PopupException as e:
+            assert_equal(e.message, 'There is no running Hive server available')
+
+          assert_equal(query_server['server_name'], 'beeswax')
+          assert_equal(query_server['server_host'], 'hive-llap-1.gethue.com')
+          assert_equal(query_server['server_port'], 10000)
+
 
-    # TODO: all the combinations in new test methods, e.g.:
-    # HIVE_DISCOVERY_LLAP_HA.get() --> True
-    # ...
+# TODO: all the combinations in new test methods, e.g.:
+# HIVE_DISCOVERY_LLAP_HA.get() --> True
+# ...

+ 6 - 2
apps/beeswax/src/beeswax/server/hive_server2_lib.py

@@ -36,7 +36,7 @@ from beeswax import conf as beeswax_conf, hive_site
 from beeswax.hive_site import hiveserver2_use_ssl
 from beeswax.conf import CONFIG_WHITELIST, LIST_PARTITIONS_LIMIT, MAX_CATALOG_SQL_ENTRIES
 from beeswax.models import Session, HiveServerQueryHandle, HiveServerQueryHistory
-from beeswax.server.dbms import Table, DataTable, QueryServerException, InvalidSessionQueryServerException
+from beeswax.server.dbms import Table, DataTable, QueryServerException, InvalidSessionQueryServerException, reset_ha
 from notebook.connectors.base import get_interpreter
 
 if sys.version_info[0] > 2:
@@ -692,7 +692,11 @@ class HiveServerClient(object):
 
     LOG.info('Opening %s thrift session for user %s' % (self.query_server['server_name'], user.username))
 
-    req = TOpenSessionReq(**kwargs)
+    try:
+      req = TOpenSessionReq(**kwargs)
+    except Exception as e:
+      if 'Connection refused' in str(e):
+        reset_ha()
     res = self._client.OpenSession(req)
     self.coordinator_host = self._client.get_coordinator_host()
     if self.coordinator_host:

+ 151 - 116
apps/metastore/src/metastore/tests.py

@@ -17,6 +17,7 @@
 # limitations under the License.
 
 from future import standard_library
+
 standard_library.install_aliases()
 from builtins import object
 import json
@@ -38,7 +39,8 @@ from useradmin.models import HuePermission, GroupPermission, User, Group
 
 from beeswax.conf import LIST_PARTITIONS_LIMIT
 from beeswax.views import collapse_whitespace
-from beeswax.test_base import make_query, wait_for_query_to_finish, verify_history, get_query_server_config, fetch_query_result_data
+from beeswax.test_base import make_query, wait_for_query_to_finish, verify_history, get_query_server_config, \
+  fetch_query_result_data
 from beeswax.models import QueryHistory
 from beeswax.server import dbms
 from beeswax.test_base import BeeswaxSampleProvider
@@ -59,23 +61,21 @@ def _make_query(client, query, submission_type="Execute",
   res = make_query(client, query, submission_type,
                    udfs, settings, resources,
                    wait, name, desc, local, is_parameterized, max, database, email_notify, **kwargs)
-
+  
   # Should be in the history if it's submitted.
   if submission_type == 'Execute':
     fragment = collapse_whitespace(smart_str(query[:20]))
     verify_history(client, fragment=fragment)
-
+  
   return res
 
 
 class TestApi():
-
   def setUp(self):
     self.client = make_logged_in_client(username="test", groupname="default", recreate=True, is_superuser=False)
-
+    
     self.user = User.objects.get(username="test")
-
-
+  
   def test_show_tables(self):
     grant_access("test", "default", "metastore")
     with patch('beeswax.server.dbms.get') as get:
@@ -91,11 +91,43 @@ class TestApi():
         ),
         server_name='hive'
       )
-
+      
       response = self.client.post('/metastore/tables/sfdc?format=json')
-
+      
       get.assert_called()
+    
+    assert_equal(response.status_code, 200)
+    data = json.loads(response.content)
+    assert_equal(data['status'], 0)
+    assert_equal(data['table_names'], ['customer', 'opportunities'])
+    assert_equal(data['tables'], [{'name': 'customer'}, {'name': 'opportunities'}])
 
+  def test_show_tables_hs2(self):
+    grant_access("test", "default", "metastore")
+    with patch('beeswax.server.dbms.get') as get:
+      with patch('beeswax.server.dbms.KazooClient') as KazooClient:
+        server_config = get_query_server_config(name='beeswax')
+        KazooClient.return_value = Mock(
+          # Bug "TypeError: expected string or buffer" if False, to add a new test case and fix
+          exists=Mock(return_value=True),
+          get_children=Mock(
+            return_value=['serverUri=hive-llap-1.gethue.com:10000;serverUri=hive-llap-2.gethue.com:10000'])
+        )
+        get.return_value = Mock(
+          get_databases=Mock(
+            return_value=['sfdc']
+          ),
+          get_database=Mock(
+            return_value={}
+          ),
+          get_tables_meta=Mock(
+            return_value=[{'name': 'customer'}, {'name': 'opportunities'}]
+          ),
+          server_name='hive'
+        )
+        response = self.client.post('/metastore/tables/sfdc?format=json')
+        get.assert_called()
+  
     assert_equal(response.status_code, 200)
     data = json.loads(response.content)
     assert_equal(data['status'], 0)
@@ -106,51 +138,53 @@ class TestApi():
 class TestMetastoreWithHadoop(BeeswaxSampleProvider):
   requires_hadoop = True
   integration = True
-
+  
   def setUp(self):
     user = User.objects.get(username='test')
     self.db = dbms.get(user, get_query_server_config())
-
+    
     add_permission("test", "test", "write", "metastore")
-
+  
   def test_basic_flow(self):
     # Default database should exist
     response = self.client.get("/metastore/databases")
     assert_true(self.db_name in response.context[0]["databases"])
-
+    
     # Table should have been created
     response = self.client.get("/metastore/tables/")
     assert_equal(200, response.status_code)
-
+    
     # Switch databases
     response = self.client.get("/metastore/tables/%s?format=json" % self.db_name)
     data = json.loads(response.content)
     assert_true('name' in data["tables"][0])
     assert_true("test" in data["table_names"])
-
+    
     # Should default to "default" database
     response = self.client.get("/metastore/tables/not_there")
     assert_equal(200, response.status_code)
-
+    
     # And have detail
     response = self.client.post("/metastore/table/%s/test/?format=json" % self.db_name, {'format': 'json'})
     data = json.loads(response.content)
     assert_true("foo" in [col['name'] for col in data['cols']])
     assert_true("SerDe Library:" in [prop['col_name'] for prop in data['properties']], data)
-
+    
     # Remember the number of history items. Use a generic fragment 'test' to pass verification.
     history_cnt = verify_history(self.client, fragment='test')
-
+    
     # Show table data.
     response = self.client.get("/metastore/table/%s/test/read" % self.db_name, follow=True)
-    response = self.client.get(reverse("beeswax:api_watch_query_refresh_json", kwargs={'id': response.context[0]['query'].id}), follow=True)
+    response = self.client.get(
+      reverse("beeswax:api_watch_query_refresh_json", kwargs={'id': response.context[0]['query'].id}), follow=True)
     response = wait_for_query_to_finish(self.client, response, max=30.0)
     # Note that it may not return all rows at once. But we expect at least 10.
     results = fetch_query_result_data(self.client, response)
     assert_true(len(results['results']) > 0)
     # This should NOT go into the query history.
-    assert_equal(verify_history(self.client, fragment='test'), history_cnt, 'Implicit queries should not be saved in the history')
-
+    assert_equal(verify_history(self.client, fragment='test'), history_cnt,
+                 'Implicit queries should not be saved in the history')
+  
   def test_show_tables(self):
     hql = """
         CREATE TABLE test_show_tables_1 (a int) COMMENT 'Test for show_tables';
@@ -159,7 +193,7 @@ class TestMetastoreWithHadoop(BeeswaxSampleProvider):
       """
     resp = _make_query(self.client, hql, database=self.db_name)
     resp = wait_for_query_to_finish(self.client, resp, max=30.0)
-
+    
     # Table should have been created
     response = self.client.get("/metastore/tables/%s?filter=show_tables&format=json" % self.db_name)
     assert_equal(200, response.status_code)
@@ -168,14 +202,14 @@ class TestMetastoreWithHadoop(BeeswaxSampleProvider):
     assert_true('name' in data["tables"][0])
     assert_true('comment' in data["tables"][0])
     assert_true('type' in data["tables"][0])
-
+    
     hql = """
         CREATE TABLE test_show_tables_4 (a int) COMMENT 'Test for show_tables';
         CREATE TABLE test_show_tables_5 (a int) COMMENT 'Test for show_tables';
       """
     resp = _make_query(self.client, hql, database=self.db_name)
     resp = wait_for_query_to_finish(self.client, resp, max=30.0)
-
+    
     # Table should have been created
     response = self.client.get("/metastore/tables/%s?filter=show_tables&format=json" % self.db_name)
     assert_equal(200, response.status_code)
@@ -184,42 +218,43 @@ class TestMetastoreWithHadoop(BeeswaxSampleProvider):
     assert_true('name' in data["tables"][0])
     assert_true('comment' in data["tables"][0])
     assert_true('type' in data["tables"][0])
-
+    
     hql = """
         CREATE INDEX test_index ON TABLE test_show_tables_1 (a) AS 'COMPACT' WITH DEFERRED REBUILD;
       """
     resp = _make_query(self.client, hql, wait=True, local=False, max=30.0, database=self.db_name)
-
+    
     # By default, index table should not appear in show tables view
     response = self.client.get("/metastore/tables/%s?format=json" % self.db_name)
     assert_equal(200, response.status_code)
     data = json.loads(response.content)
     assert_false('test_index' in data['tables'])
-
+  
   def test_describe_view(self):
     resp = self.client.post('/metastore/table/%s/myview' % self.db_name, data={'format': 'json'})
     assert_equal(200, resp.status_code, resp.content)
     data = json.loads(resp.content)
     assert_true(data['is_view'])
     assert_equal("myview", data['name'])
-
+  
   def test_describe_partitions(self):
     response = self.client.post("/metastore/table/%s/test_partitions" % self.db_name, data={'format': 'json'})
     data = json.loads(response.content)
     assert_equal(2, len(data['partition_keys']), data)
-
-    response = self.client.post("/metastore/table/%s/test_partitions/partitions" % self.db_name, data={'format': 'json'}, follow=True)
+    
+    response = self.client.post("/metastore/table/%s/test_partitions/partitions" % self.db_name,
+                                data={'format': 'json'}, follow=True)
     data = json.loads(response.content)
     partition_columns = [col for cols in data['partition_values_json'] for col in cols['columns']]
     assert_true("baz_one" in partition_columns)
     assert_true('12345' in partition_columns, partition_columns)
     assert_true("baz_foo" in partition_columns)
     assert_true('67890' in partition_columns)
-
+    
     # Not partitioned
     response = self.client.get("/metastore/table/%s/test/partitions" % self.db_name, follow=True)
     assert_true("is not partitioned." in response.content)
-
+  
   def test_describe_partitioned_table_with_limit(self):
     # We have 2 partitions in the test table
     finish = LIST_PARTITIONS_LIMIT.set_for_testing("1")
@@ -229,7 +264,7 @@ class TestMetastoreWithHadoop(BeeswaxSampleProvider):
       assert_equal(1, len(partition_values_json))
     finally:
       finish()
-
+    
     finish = LIST_PARTITIONS_LIMIT.set_for_testing("3")
     try:
       response = self.client.get("/metastore/table/%s/test_partitions/partitions" % self.db_name)
@@ -237,40 +272,43 @@ class TestMetastoreWithHadoop(BeeswaxSampleProvider):
       assert_equal(2, len(partition_values_json))
     finally:
       finish()
-
+  
   def test_read_partitions(self):
     if not is_live_cluster():
       raise SkipTest
-
+    
     partition_spec = "baz='baz_one',boom=12345"
-    response = self.client.get("/metastore/table/%s/test_partitions/partitions/%s/read" % (self.db_name, partition_spec), follow=True)
-    response = self.client.get(reverse("beeswax:api_watch_query_refresh_json", kwargs={'id': response.context[0]['query'].id}), follow=True)
+    response = self.client.get(
+      "/metastore/table/%s/test_partitions/partitions/%s/read" % (self.db_name, partition_spec), follow=True)
+    response = self.client.get(
+      reverse("beeswax:api_watch_query_refresh_json", kwargs={'id': response.context[0]['query'].id}), follow=True)
     response = wait_for_query_to_finish(self.client, response, max=30.0)
     results = fetch_query_result_data(self.client, response)
     assert_true(len(results['results']) > 0, results)
-
+  
   def test_browse_partition(self):
     partition_spec = "baz='baz_one',boom=12345"
-    response = self.client.get("/metastore/table/%s/test_partitions/partitions/%s/browse" % (self.db_name, partition_spec), follow=True)
+    response = self.client.get(
+      "/metastore/table/%s/test_partitions/partitions/%s/browse" % (self.db_name, partition_spec), follow=True)
     if is_live_cluster():
       path = '/user/hive/warehouse/%s.db/test_partitions/baz=baz_one/boom=12345' % self.db_name
     else:
       path = '/user/hive/warehouse/test_partitions/baz=baz_one/boom=12345'
     filebrowser_path = urllib.parse.unquote(reverse("filebrowser:filebrowser.views.view", kwargs={'path': path}))
     assert_equal(response.request['PATH_INFO'], filebrowser_path)
-
+  
   def test_drop_partition(self):
     # Create partition first
     partition_spec = "baz='baz_drop',boom=54321"
     hql = 'ALTER TABLE `%s`.`test_partitions` ADD IF NOT EXISTS PARTITION (%s);' % (self.db_name, partition_spec)
     resp = _make_query(self.client, hql, database=self.db_name)
     wait_for_query_to_finish(self.client, resp, max=30.0)
-
+    
     # Assert partition exists
     response = self.client.get("/metastore/table/%s/test_partitions/partitions" % self.db_name, {'format': 'json'})
     data = json.loads(response.content)
     assert_true("baz_drop" in [part['columns'][0] for part in data['partition_values_json']], data)
-
+    
     # Drop partition
     self.client.post(
       "/metastore/table/%s/test_partitions/partitions/drop" % self.db_name,
@@ -285,7 +323,7 @@ class TestMetastoreWithHadoop(BeeswaxSampleProvider):
     response = self.client.get("/metastore/table/%s/test_partitions/partitions" % self.db_name, {'format': 'json'})
     data = json.loads(response.content)
     assert_false("baz_drop" in [part['columns'][0] for part in data['partition_values_json']], data)
-
+  
   def test_drop_multi_tables(self):
     hql = """
       CREATE TABLE test_drop_1 (a int);
@@ -294,7 +332,7 @@ class TestMetastoreWithHadoop(BeeswaxSampleProvider):
     """
     resp = _make_query(self.client, hql, database=self.db_name)
     resp = wait_for_query_to_finish(self.client, resp, max=30.0)
-
+    
     # Drop them
     resp = self.client.get('/metastore/tables/drop/%s' % self.db_name, follow=True)
     assert_true('want to delete' in resp.content, resp.content)
@@ -303,7 +341,7 @@ class TestMetastoreWithHadoop(BeeswaxSampleProvider):
       {u'table_selection': [u'test_drop_1', u'test_drop_2', u'test_drop_3'], 'is_embeddable': True}
     )
     assert_equal(resp.status_code, 302)
-
+  
   def test_drop_multi_tables_with_skip_trash(self):
     hql = """
       CREATE TABLE test_drop_multi_tables_with_skip_trash_1 (a int);
@@ -312,7 +350,7 @@ class TestMetastoreWithHadoop(BeeswaxSampleProvider):
     """
     resp = _make_query(self.client, hql, database=self.db_name)
     resp = wait_for_query_to_finish(self.client, resp, max=30.0)
-
+    
     # Drop them
     resp = self.client.get('/metastore/tables/drop/%s' % self.db_name, follow=True)
     assert_true('want to delete' in resp.content, resp.content)
@@ -320,25 +358,25 @@ class TestMetastoreWithHadoop(BeeswaxSampleProvider):
       '/metastore/tables/drop/%s' % self.db_name,
       {
         u'table_selection': [u'test_drop_multi_tables_with_skip_trash_1',
-          u'test_drop_multi_tables_with_skip_trash_2', u'test_drop_multi_tables_with_skip_trash_3'],
+                             u'test_drop_multi_tables_with_skip_trash_2', u'test_drop_multi_tables_with_skip_trash_3'],
         u'skip_trash': u'on',
         'is_embeddable': True
       }
     )
     assert_equal(resp.status_code, 302)
-
+    
     response = self.client.get("/metastore/tables/%s?format=json" % self.db_name)
     assert_equal(200, response.status_code)
     data = json.loads(response.content)
     assert_false('test_drop_multi_tables_with_skip_trash_1' in data['tables'])
     assert_false('test_drop_multi_tables_with_skip_trash_2' in data['tables'])
     assert_false('test_drop_multi_tables_with_skip_trash_3' in data['tables'])
-
+  
   def test_drop_multi_databases(self):
     db1 = '%s_test_drop_1' % self.db_name
     db2 = '%s_test_drop_2' % self.db_name
     db3 = '%s_test_drop_3' % self.db_name
-
+    
     try:
       hql = """
         CREATE DATABASE %(db1)s;
@@ -347,55 +385,59 @@ class TestMetastoreWithHadoop(BeeswaxSampleProvider):
       """ % {'db1': db1, 'db2': db2, 'db3': db3}
       resp = _make_query(self.client, hql)
       resp = wait_for_query_to_finish(self.client, resp, max=30.0)
-
+      
       # Add a table to db1
       hql = "CREATE TABLE " + "`" + db1 + "`." + "`test_drop_1` (a int);"
       resp = _make_query(self.client, hql, database=db1)
       resp = wait_for_query_to_finish(self.client, resp, max=30.0)
       assert_equal(resp.status_code, 200)
-
+      
       # Drop them
       resp = self.client.get('/metastore/databases/drop', follow=True)
       assert_true('want to delete' in resp.content, resp.content)
-      resp = self.client.post('/metastore/databases/drop', {u'database_selection': [db1, db2, db3], 'is_embeddable': True})
+      resp = self.client.post('/metastore/databases/drop',
+                              {u'database_selection': [db1, db2, db3], 'is_embeddable': True})
       assert_equal(resp.status_code, 302)
     finally:
       make_query(self.client, 'DROP DATABASE IF EXISTS %(db)s' % {'db': db1}, wait=True)
       make_query(self.client, 'DROP DATABASE IF EXISTS %(db)s' % {'db': db2}, wait=True)
       make_query(self.client, 'DROP DATABASE IF EXISTS %(db)s' % {'db': db3}, wait=True)
-
-
+  
   def test_load_data(self):
     """
     Test load data queries.
     These require Hadoop, because they ask the metastore
     about whether a table is partitioned.
     """
-
+    
     # Check that view works
     resp = self.client.get("/metastore/table/%s/test/load" % self.db_name, follow=True)
     assert_true('Path' in resp.content)
-
+    
     data_dir = '%(prefix)s/tmp' % {'prefix': self.cluster.fs_prefix}
     data_path = data_dir + '/foo'
     self.cluster.fs.mkdir(data_dir)
     self.cluster.fs.create(data_path, data='123')
-
+    
     # Try the submission
-    response = self.client.post("/metastore/table/%s/test/load" % self.db_name, {'path': data_path, 'overwrite': True}, follow=True)
+    response = self.client.post("/metastore/table/%s/test/load" % self.db_name, {'path': data_path, 'overwrite': True},
+                                follow=True)
     data = json.loads(response.content)
     query = QueryHistory.objects.get(id=data['query_history_id'])
-
+    
     assert_equal_mod_whitespace(
-      "LOAD DATA INPATH '%(data_path)s' OVERWRITE INTO TABLE `%(db)s`.`test`" % {'data_path': data_path, 'db': self.db_name}, query.query
+      "LOAD DATA INPATH '%(data_path)s' OVERWRITE INTO TABLE `%(db)s`.`test`" % {'data_path': data_path,
+                                                                                 'db': self.db_name}, query.query
     )
-
-    resp = self.client.post("/metastore/table/%s/test/load" % self.db_name, {'path': data_path, 'overwrite': False}, follow=True)
+    
+    resp = self.client.post("/metastore/table/%s/test/load" % self.db_name, {'path': data_path, 'overwrite': False},
+                            follow=True)
     query = QueryHistory.objects.latest('id')
     assert_equal_mod_whitespace(
-      "LOAD DATA INPATH '%(data_path)s' INTO TABLE `%(db)s`.`test`" % {'data_path': data_path, 'db': self.db_name}, query.query
+      "LOAD DATA INPATH '%(data_path)s' INTO TABLE `%(db)s`.`test`" % {'data_path': data_path, 'db': self.db_name},
+      query.query
     )
-
+    
     # Try it with partitions
     resp = self.client.post(
       "/metastore/table/%s/test_partitions/load" % self.db_name,
@@ -409,96 +451,95 @@ class TestMetastoreWithHadoop(BeeswaxSampleProvider):
         'data_path': data_path, 'db': self.db_name
       }
     )
-
-
+  
   def test_has_write_access_frontend(self):
-    client = make_logged_in_client(username='write_access_frontend', groupname='write_access_frontend', is_superuser=False)
+    client = make_logged_in_client(username='write_access_frontend', groupname='write_access_frontend',
+                                   is_superuser=False)
     grant_access("write_access_frontend", "write_access_frontend", "metastore")
     user = User.objects.get(username='write_access_frontend')
-
+    
     def check(client, assertz):
       response = client.get("/metastore/databases")
       assertz("Drop</button>" in response.content, response.content)
       assertz("Create a new database" in response.content, response.content)
-
+      
       response = client.get("/metastore/tables/")
       assertz("Drop</button>" in response.content, response.content)
       assertz("Create a new table" in response.content, response.content)
-
+    
     check(client, assert_false)
-
+    
     # Add access
     group, created = Group.objects.get_or_create(name='write_access_frontend')
     perm, created = HuePermission.objects.get_or_create(app='metastore', action='write')
     GroupPermission.objects.get_or_create(group=group, hue_permission=perm)
-
+    
     check(client, assert_true)
-
-
+  
   def test_has_write_access_backend(self):
-    client = make_logged_in_client(username='write_access_backend', groupname='write_access_backend', is_superuser=False)
+    client = make_logged_in_client(username='write_access_backend', groupname='write_access_backend',
+                                   is_superuser=False)
     grant_access("write_access_backend", "write_access_backend", "metastore")
     grant_access("write_access_backend", "write_access_backend", "beeswax")
     user = User.objects.get(username='write_access_backend')
-
+    
     # Only fails if we were using Sentry and won't allow SELECT to user
     resp = _make_query(client, 'CREATE TABLE test_perm_1 (a int);', database=self.db_name)
     resp = wait_for_query_to_finish(client, resp, max=30.0)
-
+    
     def check(client, http_codes):
       resp = client.get('/metastore/tables/drop/%s' % self.db_name)
       assert_true(resp.status_code in http_codes, resp.content)
-
+      
       resp = client.post('/metastore/tables/drop/%s' % self.db_name, {u'table_selection': [u'test_perm_1']})
       assert_true(resp.status_code in http_codes, resp.content)
-
-    check(client, [301]) # Denied
-
+    
+    check(client, [301])  # Denied
+    
     # Add access
     group, created = Group.objects.get_or_create(name='write_access_backend')
     perm, created = HuePermission.objects.get_or_create(app='metastore', action='write')
     GroupPermission.objects.get_or_create(group=group, hue_permission=perm)
-
-    check(client, [200, 302]) # Ok
-
-
+    
+    check(client, [200, 302])  # Ok
+  
   def test_alter_database(self):
     resp = self.client.post(reverse("metastore:get_database_metadata", kwargs={'database': self.db_name}))
     json_resp = json.loads(resp.content)
     assert_true('data' in json_resp, json_resp)
     assert_true('parameters' in json_resp['data'], json_resp)
     assert_false('message=After Alter' in json_resp['data']['parameters'], json_resp)
-
+    
     # Alter message
     resp = self.client.post(reverse("metastore:alter_database", kwargs={'database': self.db_name}),
                             {'properties': json.dumps({'message': 'After Alter'})})
     json_resp = json.loads(resp.content)
     assert_equal(0, json_resp['status'], json_resp)
     assert_equal('{message=After Alter}', json_resp['data']['parameters'], json_resp)
-
-
+  
   def test_alter_table(self):
-    resp = _make_query(self.client, "CREATE TABLE test_alter_table (a int) COMMENT 'Before Alter';", database=self.db_name)
+    resp = _make_query(self.client, "CREATE TABLE test_alter_table (a int) COMMENT 'Before Alter';",
+                       database=self.db_name)
     resp = wait_for_query_to_finish(self.client, resp, max=30.0)
-
+    
     resp = self.client.get('/metastore/table/%s/test_alter_table' % self.db_name)
     assert_true('test_alter_table', resp.content)
     assert_true('Before Alter', resp.content)
-
+    
     # Alter name
     resp = self.client.post(reverse("metastore:alter_table",
                                     kwargs={'database': self.db_name, 'table': 'test_alter_table'}),
                             {'new_table_name': 'table_altered'})
     json_resp = json.loads(resp.content)
     assert_equal('table_altered', json_resp['data']['name'], json_resp)
-
+    
     # Alter comment
     resp = self.client.post(reverse("metastore:alter_table",
                                     kwargs={'database': self.db_name, 'table': 'table_altered'}),
                             {'comment': 'After Alter'})
     json_resp = json.loads(resp.content)
     assert_equal('After Alter', json_resp['data']['comment'], json_resp)
-
+    
     # Invalid table name returns error response
     resp = self.client.post(reverse("metastore:alter_table",
                                     kwargs={'database': self.db_name, 'table': 'table_altered'}),
@@ -506,26 +547,26 @@ class TestMetastoreWithHadoop(BeeswaxSampleProvider):
     json_resp = json.loads(resp.content)
     assert_equal(1, json_resp['status'], json_resp)
     assert_true('Failed to alter table' in json_resp['data'], json_resp)
-
-
+  
   def test_alter_column(self):
     resp = _make_query(self.client, 'CREATE TABLE test_alter_column (before_alter int);', database=self.db_name)
     resp = wait_for_query_to_finish(self.client, resp, max=30.0)
-
+    
     resp = self.client.get('/metastore/table/%s/test_alter_column' % self.db_name)
     assert_true('before_alter', resp.content)
     assert_true('int', resp.content)
-
+    
     # Alter name, type and comment
     resp = self.client.post(
       reverse("metastore:alter_column", kwargs={'database': self.db_name, 'table': 'test_alter_column'}),
-      {'column': 'before_alter', 'new_column_name': 'after_alter', 'new_column_type': 'string', 'comment': 'alter comment'}
+      {'column': 'before_alter', 'new_column_name': 'after_alter', 'new_column_type': 'string',
+       'comment': 'alter comment'}
     )
     json_resp = json.loads(resp.content)
     assert_equal('after_alter', json_resp['data']['name'], json_resp)
     assert_equal('string', json_resp['data']['type'], json_resp)
     assert_equal('alter comment', json_resp['data']['comment'], json_resp)
-
+    
     # Invalid column type returns error response
     resp = self.client.post(reverse("metastore:alter_column",
                                     kwargs={'database': self.db_name, 'table': 'test_alter_column'}),
@@ -536,7 +577,6 @@ class TestMetastoreWithHadoop(BeeswaxSampleProvider):
 
 
 class TestParser(object):
-
   def test_parse_simple(self):
     name = 'simple'
     type = 'string'
@@ -544,8 +584,7 @@ class TestParser(object):
     column = {'name': name, 'type': type, 'comment': comment}
     parse_tree = parser.parse_column(name, type, comment)
     assert_equal(parse_tree, column)
-
-
+  
   def test_parse_varchar(self):
     name = 'varchar'
     type = 'varchar(1000)'
@@ -553,8 +592,7 @@ class TestParser(object):
     column = {'name': name, 'type': type, 'comment': comment}
     parse_tree = parser.parse_column(name, type, comment)
     assert_equal(parse_tree, column)
-
-
+  
   def test_parse_decimal(self):
     name = 'simple'
     type = 'decimal(12,2)'
@@ -562,8 +600,7 @@ class TestParser(object):
     column = {'name': name, 'type': type, 'comment': comment}
     parse_tree = parser.parse_column(name, type, comment)
     assert_equal(parse_tree, column)
-
-
+  
   def test_parse_array(self):
     name = 'array'
     type = 'array<string>'
@@ -571,8 +608,7 @@ class TestParser(object):
     column = {'name': name, 'type': 'array', 'comment': comment, 'item': {'type': 'string'}}
     parse_tree = parser.parse_column(name, type, comment)
     assert_equal(parse_tree, column)
-
-
+  
   def test_parse_map(self):
     name = 'map'
     type = 'map<string,int>'
@@ -580,8 +616,7 @@ class TestParser(object):
     column = {'name': name, 'type': 'map', 'comment': comment, 'key': {'type': 'string'}, 'value': {'type': 'int'}}
     parse_tree = parser.parse_column(name, type, comment)
     assert_equal(parse_tree, column)
-
-
+  
   def test_parse_struct(self):
     name = 'struct'
     type = 'struct<name:string,age:int>'
@@ -592,8 +627,7 @@ class TestParser(object):
     }
     parse_tree = parser.parse_column(name, type, comment)
     assert_equal(parse_tree, column)
-
-
+  
   def test_parse_nested(self):
     name = 'nested'
     type = 'array<struct<name:string,age:int>>'
@@ -604,11 +638,11 @@ class TestParser(object):
     }
     parse_tree = parser.parse_column(name, type, comment)
     assert_equal(parse_tree, column)
-
+  
   def test_parse_nested_with_array(self):
     name = 'nested'
     type = ('struct<fieldname1:bigint,fieldname2:int,fieldname3:int,fieldname4:array<bigint>,'
-      'fieldname5:bigint,fieldname6:array<struct<array_elem:string>>,fieldname7:string>')
+            'fieldname5:bigint,fieldname6:array<struct<array_elem:string>>,fieldname7:string>')
     comment = 'test_parse_nested'
     column = {
       'comment': 'test_parse_nested',
@@ -618,7 +652,8 @@ class TestParser(object):
         {'type': 'int', 'name': 'fieldname3'},
         {'item': {'type': 'bigint'}, 'type': 'array', 'name': 'fieldname4'},
         {'type': 'bigint', 'name': 'fieldname5'},
-        {'item': {'fields': [{'type': 'string', 'name': 'array_elem'}], 'type': 'struct'}, 'type': 'array', 'name': 'fieldname6'},
+        {'item': {'fields': [{'type': 'string', 'name': 'array_elem'}], 'type': 'struct'}, 'type': 'array',
+         'name': 'fieldname6'},
         {'type': 'string', 'name': 'fieldname7'}
       ],
       'type': 'struct', 'name': 'nested'

+ 22 - 9
desktop/libs/notebook/src/notebook/connectors/hiveserver2.py

@@ -62,7 +62,7 @@ try:
   from beeswax.design import hql_query
   from beeswax.models import QUERY_TYPES, HiveServerQueryHandle, HiveServerQueryHistory, QueryHistory, Session
   from beeswax.server import dbms
-  from beeswax.server.dbms import get_query_server_config, QueryServerException
+  from beeswax.server.dbms import get_query_server_config, QueryServerException, reset_ha
   from beeswax.views import parse_out_jobs, parse_out_queries
 except ImportError as e:
   LOG.warning('Hive and HiveServer2 interfaces are not enabled: %s' % e)
@@ -101,6 +101,8 @@ def query_error_handler(func):
       message = force_unicode(str(e))
       if 'timed out' in message:
         raise OperationTimeout(e)
+      elif 'Could not connect to any' in message:
+        reset_ha()
       else:
         raise QueryError(message)
     except QueryServerException as e:
@@ -181,17 +183,28 @@ class HS2Api(Api):
   def create_session(self, lang='hive', properties=None):
     application = 'beeswax' if lang == 'hive' or lang == 'llap' else lang
 
-    if has_session_pool():
-      session = Session.objects.get_tez_session(self.user, application, MAX_NUMBER_OF_SESSIONS.get())
-    elif not has_multiple_sessions():
-      session = Session.objects.get_session(self.user, application=application)
-    else:
-      session = None
-
+    try:
+      if has_session_pool():
+        session = Session.objects.get_tez_session(self.user, application, MAX_NUMBER_OF_SESSIONS.get())
+      elif not has_multiple_sessions():
+        session = Session.objects.get_session(self.user, application=application)
+      else:
+        session = None
+    except Exception as e:
+      if 'Connection refused' in str(e) or 'Name or service not known' in str(e):
+        LOG.exception('Connection being refused or service is not available in either session or in multiple sessions'
+                      '- HA failover')
+        reset_ha()
+        
     reuse_session = session is not None
     if not reuse_session:
       db = dbms.get(self.user, query_server=get_query_server_config(name=lang, connector=self.interpreter))
-      session = db.open_session(self.user)
+      try:
+        session = db.open_session(self.user)
+      except Exception as e:
+        if 'Connection refused' in str(e) or 'Name or service not known' in str(e):
+          LOG.exception('Connection being refused or service is not available in reuse session - HA failover')
+          reset_ha()
 
     response = {
       'type': lang,