def _initialize_gateway(gateway_address):
        (host, port) = gateway_address

        callback_params = CallbackServerParameters(address=host, port=0)

        gateway = JavaGateway(GatewayClient(address=host, port=port),
                              start_callback_server=True,
                              auto_convert=True,
                              callback_server_parameters=callback_params)
        try:
            java_import(gateway.jvm, "org.apache.spark.SparkEnv")
            java_import(gateway.jvm, "org.apache.spark.SparkConf")
            java_import(gateway.jvm, "org.apache.spark.api.java.*")
            java_import(gateway.jvm, "org.apache.spark.api.python.*")
            java_import(gateway.jvm, "org.apache.spark.mllib.api.python.*")
            java_import(gateway.jvm, "org.apache.spark.sql.*")
            java_import(gateway.jvm, "org.apache.spark.sql.hive.*")
            java_import(gateway.jvm, "scala.Tuple2")
            java_import(gateway.jvm, "scala.collection.immutable.List")
        except Py4JError as e:
            log_error('Error while initializing java gateway: {}'.format(e))
            gateway.close()
            return None

        return gateway

class TypeConversionTest(unittest.TestCase):
    def setUp(self):
        self.p = start_example_app_process()
        # This is to ensure that the server is started before connecting to it!
        time.sleep(1)
        self.gateway = JavaGateway()

    def tearDown(self):
        safe_shutdown(self)
        self.p.join()

    def testLongInt(self):
        ex = self.gateway.getNewExample()
        self.assertEqual(1, ex.method7(1234))
        self.assertEqual(4, ex.method7(2147483648))
        self.assertEqual(4, ex.method7(long(2147483648)))
        self.assertEqual(long(4), ex.method8(3))
        self.assertEqual(4, ex.method8(3))
        self.assertEqual(long(4), ex.method8(long(3)))
        self.assertEqual(long(4), ex.method9(long(3)))

    def testBigDecimal(self):
        ex = self.gateway.getNewExample()
        self.assertEqual(Decimal("2147483.647"), ex.method10(2147483647, 3))
        self.assertEqual(Decimal("-13.456"), ex.method10(Decimal("-14.456")))

    def testGC(self):
        with gateway_example_app_process("nomem"):
            # This will only work with some JVM.
            gateway = JavaGateway(
                callback_server_parameters=CallbackServerParameters())
            sleep()
            example = gateway.entry_point.getNewExample()
            impl = IHelloImpl()
            self.assertEqual("This is Hello!", example.callHello(impl))
            self.assertEqual(
                "This is Hello;\n10MyMy!\n;",
                example.callHello2(impl))
            self.assertEqual(2, len(gateway.gateway_property.pool))

            # Make sure that finalizers do not block
            impl2 = IHelloImpl()
            self.assertEqual("This is Hello!", example.callHello(impl2))
            self.assertEqual(3, len(gateway.gateway_property.pool))

            gateway.jvm.java.lang.System.gc()

            # Leave time for sotimeout
            sleep(3)
            # Make sure the three objects have not been removed from the pool
            # because the Java side should not send gc request.
            self.assertEqual(len(gateway.gateway_property.pool), 3)
            gateway.shutdown()

    def testBadRetryFromJava(self):
        """Should not retry from Java to Python.
        Similar use case as testBadRetry, but from Java: Java calls a long
        Python operation.

        If there is a bug, Java will call Python, then read will fail, then it
        will call Python again.

        If there is no bug, Java will call Python, read will fail, then Java
        will raise an Exception that will be received as a Py4JError on the
        Python side.
        """
        self.p = start_short_timeout_app_process()
        gateway = JavaGateway(
            callback_server_parameters=CallbackServerParameters())
        try:
            operator = WaitOperator(0.5)
            opExample = gateway.jvm.py4j.examples.OperatorExample()

            opExample.randomBinaryOperator(operator)
            self.fail(
                "Should never retry once the first command went through."
                " number of calls made: {0}".format(operator.callCount))
        except Py4JJavaError:
            self.assertTrue(True)
        finally:
            gateway.shutdown()
            self.p.join()

 def testCallbackServer(self):
     # A close is required to stop the thread.
     gateway = JavaGateway(
         callback_server_parameters=CallbackServerParameters())
     gateway.close()
     self.assertTrue(True)
     sleep(2)

class Test(unittest.TestCase):
    def setUp(self):
#        logger = logging.getLogger("py4j")
#        logger.setLevel(logging.DEBUG)
#        logger.addHandler(logging.StreamHandler())
        self.p = start_example_app_process()
        time.sleep(0.5)
        self.gateway = JavaGateway()

    def tearDown(self):
        self.p.terminate()
        self.gateway.shutdown()
        time.sleep(0.5)

    def equal_maps(self, m1, m2):
        if len(m1) == len(m2):
            equal = True
            for k in m1:
                equal = m1[k] == m2[k]
                if not equal:
                    break
            return equal
        else:
            return False

    def testMap(self):
        dp0 = {}
        dp = get_map()
        dj = self.gateway.jvm.java.util.HashMap()
        self.equal_maps(dj, dp0)
        dj["a"] = 1
        dj["b"] = 2.0
        dj["c"] = "z"
        self.equal_maps(dj, dp)

        del(dj["a"])
        del(dp["a"])

        dj2 = self.gateway.jvm.java.util.HashMap()
        dj2["b"] = 2.0
        dj2["c"] = "z"

        dj3 = self.gateway.jvm.java.util.HashMap()
        dj3["a"] = 1
        dj3["b"] = 2.0
        dj3["c"] = "z"

        self.equal_maps(dj, dp)
        self.assertTrue(dj == dj)
        self.assertTrue(dj == dj2)
        # Does not always work for some reason...
        # Probably not worth supporting for now...
        # self.assertTrue(dj < dj3)
        self.assertTrue(dj != dp)

        dps = {1: 1, 2: 2}
        djs = self.gateway.jvm.java.util.HashMap()
        djs[1] = 1
        djs[2] = 2
        self.assertEqual(str(djs), str(dps))

class StreamTest(unittest.TestCase):
    def setUp(self):
        self.p = start_example_app_process()
        self.gateway = JavaGateway()

    def tearDown(self):
        safe_shutdown(self)
        self.p.join()

    def testBinarySuccess(self):
        e = self.gateway.getNewExample()

        # not binary - just get the Java object
        v1 = e.getStream()
        self.assertTrue(
            is_instance_of(
                self.gateway, v1, "java.nio.channels.ReadableByteChannel"))

        # pull it as a binary stream
        with e.getStream.stream() as conn:
            self.assertTrue(isinstance(conn, GatewayConnectionGuard))
            expected =\
                "Lorem ipsum dolor sit amet, consectetur adipiscing elit."
            self.assertEqual(expected, smart_decode(conn.read(len(expected))))

    def testBinaryFailure(self):
        e = self.gateway.getNewExample()
        self.assertRaises(Py4JJavaError, lambda: e.getBrokenStream())
        self.assertRaises(Py4JJavaError, lambda: e.getBrokenStream.stream())

    def testNotAStream(self):
        e = self.gateway.getNewExample()
        self.assertEqual(1, e.method1())
        self.assertRaises(Py4JError, lambda: e.method1.stream())

    def testGoodRetryFromJava(self):
        """Should retry from Java to Python.
        Similar use case as testGoodRetry, but from Java: Python calls Java,
        which calls Python back two times in a row. Then python waits for a
        while. Python then calls Java, which calls Python.

        Because Python Callback server has been waiting for too much time, the
        receiving socket has closed so the call from Java to Python will fail
        on send, and Java must retry by creating a new connection
        (CallbackConnection).
        """
        self.p = start_example_app_process()
        gateway = JavaGateway(callback_server_parameters=CallbackServerParameters(read_timeout=0.250))
        try:
            operator = WaitOperator(0)
            opExample = gateway.jvm.py4j.examples.OperatorExample()
            opExample.randomBinaryOperator(operator)
            str_connection = str(list(gateway._callback_server.connections)[0])

            opExample.randomBinaryOperator(operator)
            str_connection2 = str(list(gateway._callback_server.connections)[0])

            sleep(0.5)

            opExample.randomBinaryOperator(operator)
            str_connection3 = str(list(gateway._callback_server.connections)[0])

            self.assertEqual(str_connection, str_connection2)
            self.assertNotEqual(str_connection, str_connection3)
        except Py4JJavaError:
            self.fail("Java callbackclient did not retry.")
        finally:
            gateway.shutdown()
            self.p.join()

class ThreadTest(unittest.TestCase):
    def setUp(self):
        self.p = start_example_app_process()
        gateway_client = GatewayClient()
        self.gateway = JavaGateway()
        self.gateway.set_gateway_client(gateway_client)

    def tearDown(self):
        safe_shutdown(self)
        self.p.join()

    def testStress(self):
        # Real stress test!
        # runner1 = Runner(xrange(1,10000,2),self.gateway)
        # runner2 = Runner(xrange(1000,1000000,10000), self.gateway)
        # runner3 = Runner(xrange(1000,1000000,10000), self.gateway)
        # Small stress test
        runner1 = Runner(range(1, 10000, 1000), self.gateway)
        runner2 = Runner(range(1000, 1000000, 100000), self.gateway)
        runner3 = Runner(range(1000, 1000000, 100000), self.gateway)
        runner1.start()
        runner2.start()
        runner3.start()
        runner1.join()
        runner2.join()
        runner3.join()
        self.assertTrue(runner1.ok)
        self.assertTrue(runner2.ok)
        self.assertTrue(runner3.ok)

class HelpTest(unittest.TestCase):
    def setUp(self):
        self.p = start_example_app_process()
        self.gateway = JavaGateway()

    def tearDown(self):
        safe_shutdown(self)
        self.p.join()

    def testHelpObject(self):
        ex = self.gateway.getNewExample()
        help_page = self.gateway.help(ex, short_name=True, display=False)
        self.assertGreater(len(help_page), 1)

    def testHelpObjectWithPattern(self):
        ex = self.gateway.getNewExample()
        help_page = self.gateway.help(
            ex, pattern="m*", short_name=True, display=False)
        self.assertGreater(len(help_page), 1)

    def testHelpClass(self):
        String = self.gateway.jvm.java.lang.String
        help_page = self.gateway.help(String, short_name=False, display=False)
        self.assertGreater(len(help_page), 1)
        self.assertIn("String", help_page)

class HelpTest(unittest.TestCase):
    def setUp(self):
        self.p = start_example_app_process()
        # This is to ensure that the server is started before connecting to it!
        time.sleep(1)
        self.gateway = JavaGateway()

    def tearDown(self):
        safe_shutdown(self)
        self.p.join()

    def testHelpObject(self):
        ex = self.gateway.getNewExample()
        help_page = self.gateway.help(ex, short_name=True, display=False)
        print(help_page)
        self.assertEqual(939, len(help_page))

    def testHelpObjectWithPattern(self):
        ex = self.gateway.getNewExample()
        help_page = self.gateway.help(ex, pattern='m*', short_name=True,
                display=False)
        print(help_page)
        self.assertEqual(644, len(help_page))

    def testHelpClass(self):
        String = self.gateway.jvm.java.lang.String
        help_page = self.gateway.help(String, short_name=False, display=False)
        print(help_page)
        self.assertEqual(3439, len(help_page))

def singlethread(java_classpath):
    print "Thread starting"
    
    jvm = JVM(java_classpath, dir_path)
    socket_no = self.jvm.socket_no
    gatewayclient = GatewayClient('localhost', socket_no)
    gateway = JavaGateway(gatewayclient, auto_convert=True, auto_field=True)
    sys.stderr.write("Initialized global Java gateway with pid {} in socket {}\n".format(self.jvm.pid, socket_no))

    
    gatewayclient = GatewayClient('localhost', socket_no)
    print "Gclient started"
    gateway = JavaGateway(gatewayclient, auto_convert=True, auto_field=True)
    print "Java Gateway started"
    #create a new view for the jvm
    meteor_view = gateway.new_jvm_view()
    #import required packages
    java_import(meteor_view, 'edu.cmu.meteor.scorer.*')
    #initialize the java object
    java_import(meteor_view, 'edu.cmu.meteor.util.*')
    print "Modules imported"
    #pass the language setting into the meteor configuration object
    config = meteor_view.MeteorConfiguration();
    config.setLanguage("en");
    scorer = meteor_view.MeteorScorer(config)
    print "object initialized"
    #run object function
    stats = scorer.getMeteorStats("Test sentence", "Test sentence !");
    print stats.score
    return 1

class TypeConversionTest(unittest.TestCase):
    def setUp(self):
        self.p = start_example_app_process()
        self.gateway = JavaGateway()

    def tearDown(self):
        safe_shutdown(self)
        self.p.join()

    def testLongInt(self):
        ex = self.gateway.getNewExample()
        self.assertEqual(1, ex.method7(1234))
        self.assertEqual(4, ex.method7(2147483648))
        self.assertEqual(4, ex.method7(long(2147483648)))
        self.assertEqual(long(4), ex.method8(3))
        self.assertEqual(4, ex.method8(3))
        self.assertEqual(long(4), ex.method8(long(3)))
        self.assertEqual(long(4), ex.method9(long(3)))

    def testBigDecimal(self):
        ex = self.gateway.getNewExample()
        self.assertEqual(Decimal("2147483.647"), ex.method10(2147483647, 3))
        self.assertEqual(Decimal("-13.456"), ex.method10(Decimal("-14.456")))

    def testFloatConversion(self):
        java_inf = self.gateway.jvm.java.lang.Double.parseDouble("Infinity")
        self.assertEqual(float("inf"), java_inf)
        java_inf = self.gateway.jvm.java.lang.Double.parseDouble("+Infinity")
        self.assertEqual(float("inf"), java_inf)
        java_neg_inf = self.gateway.jvm.java.lang.Double.parseDouble(
            "-Infinity")
        self.assertEqual(float("-inf"), java_neg_inf)
        java_nan = self.gateway.jvm.java.lang.Double.parseDouble("NaN")
        self.assertTrue(math.isnan(java_nan))

    def shutdown_gateway(event, gateway: JavaGateway, resource_name: str, shutdown_jvm: bool):
        if shutdown_jvm:
            gateway.shutdown()
        else:
            gateway.close()

        logger.info('Py4J gateway (%s) shut down', resource_name)

 def testProtocolSend(self):
     testConnection = TestConnection()
     gateway = JavaGateway(testConnection, False)
     e = gateway.getExample()
     self.assertEqual('c\nt\ngetExample\ne\n', testConnection.last_message)
     e.method1(1, True, 'Hello\nWorld', e, None, 1.5)
     self.assertEqual('c\no0\nmethod1\ni1\nbTrue\nsHello\\nWorld\nro0\nn\nd1.5\ne\n', testConnection.last_message)
     del(e)

class TestIntegration(unittest.TestCase):
    def setUp(self):
        #        logger = logging.getLogger("py4j")
        #        logger.setLevel(logging.DEBUG)
        #        logger.addHandler(logging.StreamHandler())
        self.p = start_example_app_process()
        self.gateway = JavaGateway(callback_server_parameters=CallbackServerParameters())

    def tearDown(self):
        safe_shutdown(self)
        self.p.join()
        sleep()

    #    Does not work when combined with other tests... because of TCP_WAIT
    def testShutdown(self):
        example = self.gateway.entry_point.getNewExample()
        impl = IHelloImpl()
        self.assertEqual("This is Hello!", example.callHello(impl))
        self.assertEqual("This is Hello;\n10MyMy!\n;", example.callHello2(impl))
        self.gateway.shutdown()
        self.assertEqual(0, len(self.gateway.gateway_property.pool))

    def testProxy(self):
        #        self.gateway.jvm.py4j.GatewayServer.turnLoggingOn()
        sleep()
        example = self.gateway.entry_point.getNewExample()
        impl = IHelloImpl()
        self.assertEqual("This is Hello!", example.callHello(impl))
        self.assertEqual("This is Hello;\n10MyMy!\n;", example.callHello2(impl))

    def testGC(self):
        # This will only work with some JVM.
        sleep()
        example = self.gateway.entry_point.getNewExample()
        impl = IHelloImpl()
        self.assertEqual("This is Hello!", example.callHello(impl))
        self.assertEqual("This is Hello;\n10MyMy!\n;", example.callHello2(impl))
        self.assertEqual(2, len(self.gateway.gateway_property.pool))
        self.gateway.jvm.java.lang.System.gc()
        sleep(1)
        self.assertTrue(len(self.gateway.gateway_property.pool) < 2)

    def testDoubleCallbackServer(self):
        try:
            self.gateway2 = JavaGateway(callback_server_parameters=CallbackServerParameters())
            self.fail()
        except Exception:
            self.assertTrue(True)

    def testMethodConstructor(self):
        sleep()
        goodAddition = GoodAddition()
        oe1 = self.gateway.jvm.py4j.examples.OperatorExample()
        # Test method
        oe1.randomBinaryOperator(goodAddition)
        # Test constructor
        oe2 = self.gateway.jvm.py4j.examples.OperatorExample(goodAddition)
        self.assertTrue(oe2 is not None)

class ProtocolTest(unittest.TestCase):
    def tearDown(self):
        # Safety check in case there was an exception...
        safe_shutdown(self)

    def testEscape(self):
        self.assertEqual("Hello\t\rWorld\n\\", unescape_new_line(
            escape_new_line("Hello\t\rWorld\n\\")))
        self.assertEqual("Hello\t\rWorld\n\\", unescape_new_line(
            escape_new_line("Hello\t\rWorld\n\\")))

    def testProtocolSend(self):
        testConnection = TestConnection()
        self.gateway = JavaGateway(testConnection, False)
        e = self.gateway.getExample()
        self.assertEqual('c\nt\ngetExample\ne\n', testConnection.last_message)
        e.method1(1, True, 'Hello\nWorld', e, None, 1.5)
        self.assertEqual(
                'c\no0\nmethod1\ni1\nbTrue\nsHello\\nWorld\nro0\nn\nd1.5\ne\n',
                testConnection.last_message)
        del(e)

    def testProtocolReceive(self):
        p = start_echo_server_process()
        time.sleep(1)
        try:
            testSocket = get_socket()
            testSocket.sendall('yo\n'.encode('utf-8'))
            testSocket.sendall('yro0\n'.encode('utf-8'))
            testSocket.sendall('yo\n'.encode('utf-8'))
            testSocket.sendall('ysHello World\n'.encode('utf-8'))
            # No extra echange (method3) because it is already cached.
            testSocket.sendall('yi123\n'.encode('utf-8'))
            testSocket.sendall('yd1.25\n'.encode('utf-8'))
            testSocket.sendall('yo\n'.encode('utf-8'))
            testSocket.sendall('yn\n'.encode('utf-8'))
            testSocket.sendall('yo\n'.encode('utf-8'))
            testSocket.sendall('ybTrue\n'.encode('utf-8'))
            testSocket.sendall('yo\n'.encode('utf-8'))
            testSocket.sendall('yL123\n'.encode('utf-8'))
            testSocket.close()
            time.sleep(1)

            self.gateway = JavaGateway(auto_field=True)
            ex = self.gateway.getNewExample()
            self.assertEqual('Hello World', ex.method3(1, True))
            self.assertEqual(123, ex.method3())
            self.assertAlmostEqual(1.25, ex.method3())
            self.assertTrue(ex.method2() is None)
            self.assertTrue(ex.method4())
            self.assertEqual(long(123), ex.method8())
            self.gateway.shutdown()

        except Exception as e:
            print('Error has occurred', e)
            print_exc()
            self.fail('Problem occurred')
        p.join()

def gateway(*args, **kwargs):
    g = JavaGateway(gateway_parameters=GatewayParameters(*args, auto_convert=True, **kwargs))
    lineSep = g.jvm.System.lineSeparator()
    try:
        yield g
        # Call a dummy method to make sure we haven't corrupted the streams
        assert lineSep == g.jvm.System.lineSeparator()
    finally:
        g.shutdown()

def gateway(*args, **kwargs):
    g = JavaGateway(gateway_parameters=GatewayParameters(*args, auto_convert=True, **kwargs))
    time = g.jvm.System.currentTimeMillis()
    try:
        yield g
        # Call a dummy method to make sure we haven't corrupted the streams
        assert time <= g.jvm.System.currentTimeMillis()
    finally:
        g.shutdown()

def check_JavaGateway():
    try:
        global app
        app = JavaGateway(python_proxy_port=port).entry_point

        if app.test() != "TESTOK":
            restart_JavaGateway()

    except Exception:
        restart_JavaGateway()