diff --git a/.github/workflows/build-test.yml b/.github/workflows/build-test.yml
index 818e8819a6e..f3a93b0c2ab 100644
--- a/.github/workflows/build-test.yml
+++ b/.github/workflows/build-test.yml
@@ -163,7 +163,7 @@ jobs:
         run: |
           touch gremlin-python/.glv
           mvn clean install -pl -:gremlin-javascript,-gremlin-dotnet,-:gremlin-dotnet-source,-:gremlin-dotnet-tests,-:gremlint -q -DskipTests -Dci
-          mvn verify -pl gremlin-python
+          mvn verify -pl gremlin-python -DincludeNeo4j
   dotnet:
     name: .NET
     timeout-minutes: 1200 # 20 minutes
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 3dca81956a8..fa945de09df 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -23,6 +23,9 @@ image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima
 [[release-3-5-2]]
 === TinkerPop 3.5.2 (Release Date: NOT OFFICIALLY RELEASED YET)
 
+* Added support for `g.Tx()` in Python.
+* Added logging in in Python.
+* Fixed shutdown cleanup issue in Python aiohttp transport layer.
 * Added a `NoSugarTranslator` translator to `PythonTranslator` which translates Gremlin queries to Python without syntactic sugar (ex `g.V().limit(1)` instead of `g.V()[0:1]`)
 * Added support for `g.Tx()` in .NET.
 * Added support for `with()` constant options to `io()`.
diff --git a/docs/src/reference/gremlin-variants.asciidoc b/docs/src/reference/gremlin-variants.asciidoc
index d107d078b18..b994237d60b 100644
--- a/docs/src/reference/gremlin-variants.asciidoc
+++ b/docs/src/reference/gremlin-variants.asciidoc
@@ -913,6 +913,37 @@ Apache TinkerPop's JVM-based Gremlin traversal machine. As such, their `apply(Tr
 the strategy is encoded in the Gremlin-Python bytecode and transmitted to the Gremlin traversal machine for
 re-construction machine-side.
 
+[[gremlin-python-transactions]]
+=== Transactions
+
+To get a full understanding of this section, it would be good to start by reading the <<transactions,Transactions>>
+section of this documentation, which discusses transactions in the general context of TinkerPop itself. This section
+builds on that content by demonstrating the transactional syntax for Javascript.
+
+[source,python]
+----
+g = traversal().withRemote(DriverRemoteConnection('ws://localhost:8182/gremlin'))
+
+# Create a Transaction.
+tx = g.tx()
+
+# Spawn a new GraphTraversalSource, binding all traversals established from it to tx.
+gtx = tx.begin()
+
+try:
+    # Execute a traversal within the transaction.
+    gtx.addV("person").property("name", "Lyndon").iterate(),
+
+    # Commit the transaction. The transaction can no longer be used and cannot be re-used.
+    # A new transaction can be spawned through g.tx().
+    # The context of g remains sessionless throughout the process.
+    gtx.commit()
+except Exception e:
+    # Rollback the transaction if an error occurs.
+    gtx.rollback()
+
+----
+
 [[gremlin-python-lambda]]
 === The Lambda Solution
 
diff --git a/docs/src/upgrade/release-3.5.x.asciidoc b/docs/src/upgrade/release-3.5.x.asciidoc
index 882c8c19826..6a4dd73c6e7 100644
--- a/docs/src/upgrade/release-3.5.x.asciidoc
+++ b/docs/src/upgrade/release-3.5.x.asciidoc
@@ -30,13 +30,20 @@ complete list of all the modifications that are part of this release.
 
 === Upgrading for Users
 
-==== tx() in .NET
+==== Tx() in .NET and tx() in Python
 
-After Javascript, .NET is now the second non-JVM variant of Gremlin to get support for
-link:https://tinkerpop.apache.org/docs/3.5.2/reference/#transactions[remote transactions]. An example of the `Tx()`
-syntax can be found in the .NET link:https://tinkerpop.apache.org/docs/3.5.2/reference/#gremlin-dotnet-transactions[Transaction Section].
+After Javascript, .NET and Python are now the second and third non-JVM variants of Gremlin to get support for
+link:https://tinkerpop.apache.org/docs/3.5.2/reference/#transactions[remote transactions].
 
-See: link:https://issues.apache.org/jira/browse/TINKERPOP-2556[TINKERPOP-2556]
+An example of the .NET `Tx()` syntax can be found in the
+link:https://tinkerpop.apache.org/docs/3.5.2/reference/#gremlin-dotnet-transactions[.NET Transaction Section].
+
+An example of the Python `tx()` syntax can be found in the
+link:https://tinkerpop.apache.org/docs/3.5.2/reference/#gremlin-dotnet-transactions[Python Transaction Section]
+
+
+See: link:https://issues.apache.org/jira/browse/TINKERPOP-2556[TINKERPOP-2556] for .NET and
+link:https://issues.apache.org/jira/browse/TINKERPOP-2555[TINKERPOP-2555] for Python
 
 ==== datetime()
 
diff --git a/gremlin-python/pom.xml b/gremlin-python/pom.xml
index 87dcaf82955..bbaa75ee852 100644
--- a/gremlin-python/pom.xml
+++ b/gremlin-python/pom.xml
@@ -29,6 +29,7 @@ limitations under the License.
         <!-- provides a way to convert maven.test.skip value to skipTests for use in skipping python tests -->
         <maven.test.skip>false</maven.test.skip>
         <skipTests>${maven.test.skip}</skipTests>
+        <TEST_TRANSACTIONS>false</TEST_TRANSACTIONS>
         <gremlin.server.dir>${project.parent.basedir}/gremlin-server</gremlin.server.dir>
     </properties>
     <build>
@@ -216,6 +217,7 @@ limitations under the License.
                                     <target>
                                         <exec executable="env/bin/python" dir="${project.build.directory}/python3"
                                               failonerror="true">
+                                            <env key="TEST_TRANSACTIONS" value="${TEST_TRANSACTIONS}"/>
                                             <env key="PYTHONPATH" value=""/>
                                             <env key="KRB5_CONFIG" value="${project.build.directory}/kdc/krb5.conf"/>
                                             <env key="KRB5CCNAME" value="${project.build.directory}/kdc/test-tkt.cc"/>
@@ -224,12 +226,14 @@ limitations under the License.
                                         <!-- radish seems to like all dependencies in place -->
                                         <exec executable="env/bin/python" dir="${project.build.directory}/python3"
                                               failonerror="true">
+                                            <env key="TEST_TRANSACTIONS" value="${TEST_TRANSACTIONS}"/>
                                             <env key="PYTHONPATH" value=""/>
                                             <arg line="setup.py install"/>
                                         </exec>
                                         <!-- run for graphson 3.0 -->
                                         <exec executable="env/bin/radish" dir="${project.build.directory}/python3"
                                               failonerror="true">
+                                            <env key="TEST_TRANSACTIONS" value="${TEST_TRANSACTIONS}"/>
                                             <env key="PYTHONPATH" value=""/>
                                             <env key="PYTHONIOENCODING" value="utf-8:surrogateescape"/>
                                             <arg line="-f dots -e -t -b ${project.build.directory}/python3/radish ${project.basedir}/../gremlin-test/features/ --user-data=&quot;serializer=application/vnd.gremlin-v3.0+json&quot;"/> <!-- -no-line-jump -->
@@ -237,6 +241,7 @@ limitations under the License.
                                         <!-- run for graphbinary 1.0 -->
                                         <exec executable="env/bin/radish" dir="${project.build.directory}/python3"
                                               failonerror="true">
+                                            <env key="TEST_TRANSACTIONS" value="${TEST_TRANSACTIONS}"/>
                                             <env key="PYTHONPATH" value=""/>
                                             <env key="PYTHONIOENCODING" value="utf-8:surrogateescape"/>
                                             <arg line="-f dots -e -t -b ${project.build.directory}/python3/radish ${project.basedir}/../gremlin-test/features/ --user-data=&quot;serializer=application/vnd.graphbinary-v1.0&quot;"/> <!-- -no-line-jump -->
@@ -260,6 +265,11 @@ limitations under the License.
                                 <artifactId>gremlin-test</artifactId>
                                 <version>${project.version}</version>
                             </dependency>
+                            <dependency>
+                                <groupId>org.apache.tinkerpop</groupId>
+                                <artifactId>neo4j-gremlin</artifactId>
+                                <version>${project.version}</version>
+                            </dependency>
                             <dependency>
                                 <groupId>org.codehaus.groovy</groupId>
                                 <artifactId>groovy-all</artifactId>
@@ -354,6 +364,118 @@ limitations under the License.
                 </plugins>
             </build>
         </profile>
+        <!--
+          This profile will include neo4j for purposes of transactional testing within Gremlin Server.
+          Tests that require neo4j specifically will be "ignored" if this profile is not turned on.
+        -->
+        <profile>
+            <id>include-neo4j</id>
+            <activation>
+                <activeByDefault>false</activeByDefault>
+                <property>
+                    <name>includeNeo4j</name>
+                </property>
+            </activation>
+            <properties>
+                <TEST_TRANSACTIONS>true</TEST_TRANSACTIONS>
+            </properties>
+            <build>
+                <plugins>
+                    <plugin>
+                        <groupId>org.codehaus.gmavenplus</groupId>
+                        <artifactId>gmavenplus-plugin</artifactId>
+                        <dependencies>
+                            <dependency>
+                                <groupId>org.neo4j</groupId>
+                                <artifactId>neo4j-tinkerpop-api-impl</artifactId>
+                                <version>0.9-3.4.0</version>
+                                <exclusions>
+                                    <exclusion>
+                                        <groupId>org.neo4j</groupId>
+                                        <artifactId>neo4j-kernel</artifactId>
+                                    </exclusion>
+                                    <exclusion>
+                                        <groupId>org.apache.commons</groupId>
+                                        <artifactId>commons-lang3</artifactId>
+                                    </exclusion>
+                                    <exclusion>
+                                        <groupId>org.apache.commons</groupId>
+                                        <artifactId>commons-text</artifactId>
+                                    </exclusion>
+                                    <exclusion>
+                                        <groupId>com.github.ben-manes.caffeine</groupId>
+                                        <artifactId>caffeine</artifactId>
+                                    </exclusion>
+                                    <exclusion>
+                                        <groupId>org.scala-lang</groupId>
+                                        <artifactId>scala-library</artifactId>
+                                    </exclusion>
+                                    <exclusion>
+                                        <groupId>org.scala-lang</groupId>
+                                        <artifactId>scala-reflect</artifactId>
+                                    </exclusion>
+                                    <exclusion>
+                                        <groupId>org.slf4j</groupId>
+                                        <artifactId>slf4j-api</artifactId>
+                                    </exclusion>
+                                    <exclusion>
+                                        <groupId>org.slf4j</groupId>
+                                        <artifactId>slf4j-nop</artifactId>
+                                    </exclusion>
+                                    <exclusion>
+                                        <groupId>org.apache.lucene</groupId>
+                                        <artifactId>lucene-core</artifactId>
+                                    </exclusion>
+                                    <exclusion>
+                                        <groupId>io.dropwizard.metrics</groupId>
+                                        <artifactId>metrics-core</artifactId>
+                                    </exclusion>
+                                    <exclusion>
+                                        <groupId>io.netty</groupId>
+                                        <artifactId>netty-all</artifactId>
+                                    </exclusion>
+                                    <exclusion>
+                                        <groupId>org.ow2.asm</groupId>
+                                        <artifactId>asm</artifactId>
+                                    </exclusion>
+                                </exclusions>
+                            </dependency>
+                            <dependency>
+                                <groupId>org.scala-lang</groupId>
+                                <artifactId>scala-library</artifactId>
+                                <version>2.11.8</version>
+                            </dependency>
+                            <dependency>
+                                <groupId>org.scala-lang</groupId>
+                                <artifactId>scala-reflect</artifactId>
+                                <version>2.11.8</version>
+                            </dependency>
+                            <dependency>
+                                <groupId>org.apache.lucene</groupId>
+                                <artifactId>lucene-core</artifactId>
+                                <version>5.5.0</version>
+                            </dependency>
+                            <dependency>
+                                <groupId>io.dropwizard.metrics</groupId>
+                                <artifactId>metrics-core</artifactId>
+                                <version>4.0.2</version>
+                            </dependency>
+                            <dependency>
+                                <groupId>org.neo4j</groupId>
+                                <artifactId>neo4j-kernel</artifactId>
+                                <version>3.4.11</version>
+                                <exclusions>
+                                    <exclusion>
+                                        <groupId>io.netty</groupId>
+                                        <artifactId>netty-all</artifactId>
+                                    </exclusion>
+                                </exclusions>
+                            </dependency>
+                        </dependencies>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
         <!--
         Provides a way to deploy the gremlinpython GLV to pypi. This cannot be part of the standard maven execution
         because pypi does not have a staging environment like sonatype for releases. As soon as the release is
diff --git a/gremlin-python/src/main/python/gremlin_python/driver/aiohttp/transport.py b/gremlin-python/src/main/python/gremlin_python/driver/aiohttp/transport.py
index a964ae6a462..2effcf1b533 100644
--- a/gremlin-python/src/main/python/gremlin_python/driver/aiohttp/transport.py
+++ b/gremlin-python/src/main/python/gremlin_python/driver/aiohttp/transport.py
@@ -16,12 +16,11 @@
 # specific language governing permissions and limitations
 # under the License.
 #
-
-
 import aiohttp
 import asyncio
 import async_timeout
 from aiohttp import ClientResponseError
+import logging
 
 from gremlin_python.driver.transport import AbstractBaseTransport
 
@@ -58,6 +57,11 @@ def __init__(self, call_from_event_loop=None, read_timeout=None, write_timeout=N
         if "ssl_options" in self._aiohttp_kwargs:
             self._aiohttp_kwargs["ssl"] = self._aiohttp_kwargs.pop("ssl_options")
 
+    def __del__(self):
+        # Close will only actually close if things are left open, so this is safe to call.
+        # Clean up any connection resources and close the event loop.
+        self.close()
+
     def connect(self, url, headers=None):
         # Inner function to perform async connect.
         async def async_connect():
@@ -116,10 +120,13 @@ async def async_read():
     def close(self):
         # Inner function to perform async close.
         async def async_close():
-            if not self._websocket.closed:
+            if self._websocket is not None and not self._websocket.closed:
                 await self._websocket.close()
-            if not self._client_session.closed:
+                self._websocket = None
+
+            if self._client_session is not None and not self._client_session.closed:
                 await self._client_session.close()
+                self._client_session = None
 
         # If the loop is not closed (connection hasn't already been closed)
         if not self._loop.is_closed():
diff --git a/gremlin-python/src/main/python/gremlin_python/driver/client.py b/gremlin-python/src/main/python/gremlin_python/driver/client.py
index 8b6a53b330b..dca193b6942 100644
--- a/gremlin-python/src/main/python/gremlin_python/driver/client.py
+++ b/gremlin-python/src/main/python/gremlin_python/driver/client.py
@@ -16,8 +16,9 @@
 # specific language governing permissions and limitations
 # under the License.
 #
+import logging
+import warnings
 from concurrent.futures import ThreadPoolExecutor
-
 from six.moves import queue
 
 from gremlin_python.driver import connection, protocol, request, serializer
@@ -31,7 +32,7 @@
     def cpu_count():
         return None
 
-__author__ = 'David M. Brown (davebshow@gmail.com)'
+__author__ = 'David M. Brown (davebshow@gmail.com), Lyndon Bauto (lyndonb@bitquilltech.com)'
 
 
 class Client:
@@ -39,8 +40,9 @@ class Client:
     def __init__(self, url, traversal_source, protocol_factory=None,
                  transport_factory=None, pool_size=None, max_workers=None,
                  message_serializer=None, username="", password="",
-                 kerberized_service="", headers=None, session="",
+                 kerberized_service="", headers=None, session=None,
                  **transport_kwargs):
+        logging.info("Creating Client with url '%s'", url)
         self._url = url
         self._headers = headers
         self._traversal_source = traversal_source
@@ -52,7 +54,7 @@ def __init__(self, url, traversal_source, protocol_factory=None,
         self._username = username
         self._password = password
         self._session = session
-        self._sessionEnabled = (session != "")
+        self._session_enabled = (session is not None and session != "")
         if transport_factory is None:
             try:
                 from gremlin_python.driver.aiohttp.transport import (
@@ -71,7 +73,7 @@ def protocol_factory(): return protocol.GremlinServerWSProtocol(
                 password=self._password,
                 kerberized_service=kerberized_service)
         self._protocol_factory = protocol_factory
-        if self._sessionEnabled:
+        if self._session_enabled:
             if pool_size is None:
                 pool_size = 1
             elif pool_size != 1:
@@ -107,8 +109,9 @@ def _fill_pool(self):
             self._pool.put_nowait(conn)
 
     def close(self):
-        if self._sessionEnabled:
+        if self._session_enabled:
             self._close_session()
+        logging.info("Closing Client with url '%s'", self._url)
         while not self._pool.empty():
             conn = self._pool.get(True)
             conn.close()
@@ -117,7 +120,7 @@ def close(self):
     def _close_session(self):
         message = request.RequestMessage(
             processor='session', op='close',
-            args={'session': self._session})
+            args={'session': str(self._session)})
         conn = self._pool.get(True)
         return conn.write(message).result()
 
@@ -129,24 +132,35 @@ def _get_connection(self):
             headers=self._headers)
 
     def submit(self, message, bindings=None, request_options=None):
-        return self.submitAsync(message, bindings=bindings, request_options=request_options).result()
+        return self.submit_async(message, bindings=bindings, request_options=request_options).result()
 
     def submitAsync(self, message, bindings=None, request_options=None):
+        warnings.warn(
+            "gremlin_python.driver.client.Client.submitAsync will be replaced by "
+            "gremlin_python.driver.client.Client.submit_async.",
+            DeprecationWarning)
+        self.submit_async(message, bindings, request_options)
+
+    def submit_async(self, message, bindings=None, request_options=None):
+        logging.debug("message '%s'", str(message))
+        args = {'gremlin': message, 'aliases': {'g': self._traversal_source}}
+        processor = ''
+        op = 'eval'
         if isinstance(message, traversal.Bytecode):
-            message = request.RequestMessage(
-                processor='traversal', op='bytecode',
-                args={'gremlin': message,
-                      'aliases': {'g': self._traversal_source}})
-        elif isinstance(message, str):
-            message = request.RequestMessage(
-                processor='', op='eval',
-                args={'gremlin': message,
-                      'aliases': {'g': self._traversal_source}})
-            if bindings:
-                message.args.update({'bindings': bindings})
-            if self._sessionEnabled:
-                message = message._replace(processor='session')
-                message.args.update({'session': self._session})
+            op = 'bytecode'
+            processor = 'traversal'
+
+        if isinstance(message, str) and bindings:
+            args['bindings'] = bindings
+
+        if self._session_enabled:
+            args['session'] = str(self._session)
+            processor = 'session'
+
+        if isinstance(message, traversal.Bytecode) or isinstance(message, str):
+            logging.debug("processor='%s', op='%s', args='%s'", str(processor), str(op), str(args))
+            message = request.RequestMessage(processor=processor, op=op, args=args)
+
         conn = self._pool.get(True)
         if request_options:
             message.args.update(request_options)
diff --git a/gremlin-python/src/main/python/gremlin_python/driver/driver_remote_connection.py b/gremlin-python/src/main/python/gremlin_python/driver/driver_remote_connection.py
index abc4a2c4747..0767022e6cb 100644
--- a/gremlin-python/src/main/python/gremlin_python/driver/driver_remote_connection.py
+++ b/gremlin-python/src/main/python/gremlin_python/driver/driver_remote_connection.py
@@ -16,24 +16,44 @@
 # specific language governing permissions and limitations
 # under the License.
 #
+import logging
 from concurrent.futures import Future
+import warnings
 
 from gremlin_python.driver import client, serializer
 from gremlin_python.driver.remote_connection import (
     RemoteConnection, RemoteTraversal)
 from gremlin_python.process.strategies import OptionsStrategy
+from gremlin_python.process.traversal import Bytecode
+import uuid
 
-__author__ = 'David M. Brown (davebshow@gmail.com)'
+__author__ = 'David M. Brown (davebshow@gmail.com), Lyndon Bauto (lyndonb@bitquilltech.com)'
 
 
 class DriverRemoteConnection(RemoteConnection):
 
-    def __init__(self, url, traversal_source = "g", protocol_factory=None,
+    def __init__(self, url, traversal_source="g", protocol_factory=None,
                  transport_factory=None, pool_size=None, max_workers=None,
                  username="", password="", kerberized_service='',
                  message_serializer=None, graphson_reader=None,
-                 graphson_writer=None, headers=None,
+                 graphson_writer=None, headers=None, session=None,
                  **transport_kwargs):
+        logging.info("Creating DriverRemoteConnection with url '%s'", str(url))
+        self.__url = url
+        self.__traversal_source = traversal_source
+        self.__protocol_factory = protocol_factory
+        self.__transport_factory = transport_factory
+        self.__pool_size = pool_size
+        self.__max_workers = max_workers
+        self.__username = username
+        self.__password = password
+        self.__kerberized_service = kerberized_service
+        self.__message_serializer = message_serializer
+        self.__graphson_reader = graphson_reader
+        self.__graphson_writer = graphson_writer
+        self.__headers = headers
+        self.__session = session
+        self.__transport_kwargs = transport_kwargs
         if message_serializer is None:
             message_serializer = serializer.GraphSONMessageSerializer(
                 reader=graphson_reader,
@@ -48,21 +68,32 @@ def __init__(self, url, traversal_source = "g", protocol_factory=None,
                                      password=password,
                                      kerberized_service=kerberized_service,
                                      headers=headers,
+                                     session=session,
                                      **transport_kwargs)
         self._url = self._client._url
         self._traversal_source = self._client._traversal_source
 
     def close(self):
+        logging.info("closing DriverRemoteConnection with url '%s'", str(self._url))
         self._client.close()
 
     def submit(self, bytecode):
+        logging.debug("submit with bytecode '%s'", str(bytecode))
         result_set = self._client.submit(bytecode, request_options=self._extract_request_options(bytecode))
         results = result_set.all().result()
         return RemoteTraversal(iter(results))
 
-    def submitAsync(self, bytecode):
+    def submitAsync(self, message, bindings=None, request_options=None):
+        warnings.warn(
+            "gremlin_python.driver.driver_remote_connection.DriverRemoteConnection.submitAsync will be replaced by "
+            "gremlin_python.driver.driver_remote_connection.DriverRemoteConnection.submit_async.",
+            DeprecationWarning)
+        self.submit_async(message, bindings, request_options)
+
+    def submit_async(self, bytecode):
+        logging.debug("submit_async with bytecode '%s'", str(bytecode))
         future = Future()
-        future_result_set = self._client.submitAsync(bytecode, request_options=self._extract_request_options(bytecode))
+        future_result_set = self._client.submit_async(bytecode, request_options=self._extract_request_options(bytecode))
 
         def cb(f):
             try:
@@ -75,6 +106,37 @@ def cb(f):
         future_result_set.add_done_callback(cb)
         return future
 
+    def is_session_bound(self):
+        return self.__session is not None
+
+    def create_session(self):
+        logging.info("Creating session based connection")
+        if self.is_session_bound():
+            raise Exception('Connection is already bound to a session - child sessions are not allowed')
+        return DriverRemoteConnection(self.__url,
+                                      traversal_source=self.__traversal_source,
+                                      protocol_factory=self.__protocol_factory,
+                                      transport_factory=self.__transport_factory,
+                                      pool_size=self.__pool_size,
+                                      max_workers=self.__max_workers,
+                                      username=self.__username,
+                                      password=self.__password,
+                                      kerberized_service=self.__kerberized_service,
+                                      message_serializer=self.__message_serializer,
+                                      graphson_reader=self.__graphson_reader,
+                                      graphson_writer=self.__graphson_writer,
+                                      headers=self.__headers,
+                                      session=uuid.uuid4(),
+                                      **self.__transport_kwargs)
+
+    def commit(self):
+        logging.info("Submitting commit graph operation.")
+        return self._client.submit(Bytecode.GraphOp.commit())
+
+    def rollback(self):
+        logging.info("Submitting rollback graph operation.")
+        return self._client.submit(Bytecode.GraphOp.rollback())
+
     @staticmethod
     def _extract_request_options(bytecode):
         options_strategy = next((x for x in bytecode.source_instructions
diff --git a/gremlin-python/src/main/python/gremlin_python/driver/protocol.py b/gremlin-python/src/main/python/gremlin_python/driver/protocol.py
index 74773b8c50e..895b4ee8128 100644
--- a/gremlin-python/src/main/python/gremlin_python/driver/protocol.py
+++ b/gremlin-python/src/main/python/gremlin_python/driver/protocol.py
@@ -16,6 +16,7 @@
 # specific language governing permissions and limitations
 # under the License.
 #
+import logging
 import abc
 import base64
 import struct
@@ -81,13 +82,13 @@ def connection_made(self, transport):
         super(GremlinServerWSProtocol, self).connection_made(transport)
 
     def write(self, request_id, request_message):
-        message = self._message_serializer.serialize_message(
-            request_id, request_message)
+        message = self._message_serializer.serialize_message(request_id, request_message)
         self._transport.write(message)
 
     def data_received(self, message, results_dict):
         # if Gremlin Server cuts off then we get a None for the message
         if message is None:
+            logging.error("Received empty message from server.")
             raise GremlinServerError({'code': 500,
                                       'message': 'Server disconnected - please try to reconnect', 'attributes': {}})
 
@@ -108,10 +109,11 @@ def data_received(self, message, results_dict):
             elif self._kerberized_service:
                 request_message = self._kerberos_received(message)
             else:
-                raise ConfigurationError(
-                    'Gremlin server requires authentication credentials in DriverRemoteConnection.'
-                    'For basic authentication provide username and password. '
-                    'For kerberos authentication provide the kerberized_service parameter.')
+                error_message = 'Gremlin server requires authentication credentials in DriverRemoteConnection. ' \
+                                'For basic authentication provide username and password. ' \
+                                'For kerberos authentication provide the kerberized_service parameter.'
+                logging.error(error_message)
+                raise ConfigurationError(error_message)
             self.write(request_id, request_message)
             data = self._transport.read()
             # Allow for auth handshake with multiple steps
@@ -127,6 +129,10 @@ def data_received(self, message, results_dict):
                 del results_dict[request_id]
             return status_code
         else:
+            # This message is going to be huge and kind of hard to read, but in the event of an error,
+            # it can provide invaluable info, so space it out appropriately.
+            logging.error("\r\nReceived error message '%s'\r\n\r\nWith results dictionary '%s'",
+                          str(message), str(results_dict))
             del results_dict[request_id]
             raise GremlinServerError(message['status'])
 
diff --git a/gremlin-python/src/main/python/gremlin_python/driver/remote_connection.py b/gremlin-python/src/main/python/gremlin_python/driver/remote_connection.py
index 667354ef941..6c5830dace2 100644
--- a/gremlin-python/src/main/python/gremlin_python/driver/remote_connection.py
+++ b/gremlin-python/src/main/python/gremlin_python/driver/remote_connection.py
@@ -19,10 +19,9 @@
 import abc
 import six
 
-from gremlin_python.driver import request
 from gremlin_python.process import traversal
 
-__author__ = 'Marko A. Rodriguez (http://markorodriguez.com)'
+__author__ = 'Marko A. Rodriguez (http://markorodriguez.com), Lyndon Bauto (lyndonb@bitquilltech.com)'
 
 
 @six.add_metaclass(abc.ABCMeta)
@@ -43,6 +42,18 @@ def traversal_source(self):
     def submit(self, bytecode):
         pass
 
+    def is_session_bound(self):
+        return False
+
+    def create_session(self):
+        raise Exception('createSession() must be implemented');
+
+    def commit(self):
+        raise Exception('commit() must be implemented')
+
+    def rollback(self):
+        raise Exception('rollback() must be implemented')
+
     def __repr__(self):
         return "remoteconnection[" + self._url + "," + self._traversal_source + "]"
 
@@ -55,7 +66,10 @@ def __init__(self, traversers):
 
 class RemoteStrategy(traversal.TraversalStrategy):
     def __init__(self, remote_connection):
-        traversal.TraversalStrategy.__init__(self)
+        # Gave this a fqcn that has a local "py:" prefix since this strategy isn't sent as bytecode to the server.
+        # this is a sort of local-only strategy that actually executes client side. not sure if this prefix is the
+        # right way to name this or not, but it should have a name to identify it.
+        traversal.TraversalStrategy.__init__(self, fqcn="py:RemoteStrategy")
         self.remote_connection = remote_connection
 
     def apply(self, traversal):
@@ -66,5 +80,4 @@ def apply(self, traversal):
 
     def apply_async(self, traversal):
         if traversal.traversers is None:
-            traversal.remote_results = self.remote_connection.submitAsync(
-                traversal.bytecode)
+            traversal.remote_results = self.remote_connection.submit_async(traversal.bytecode)
diff --git a/gremlin-python/src/main/python/gremlin_python/driver/serializer.py b/gremlin-python/src/main/python/gremlin_python/driver/serializer.py
index 6818dc9e45c..42e1aebaaf5 100644
--- a/gremlin-python/src/main/python/gremlin_python/driver/serializer.py
+++ b/gremlin-python/src/main/python/gremlin_python/driver/serializer.py
@@ -28,7 +28,7 @@
 from gremlin_python.structure.io import graphsonV2d0
 from gremlin_python.structure.io import graphsonV3d0
 
-__author__ = 'David M. Brown (davebshow@gmail.com)'
+__author__ = 'David M. Brown (davebshow@gmail.com), Lyndon Bauto (lyndonb@bitquilltech.com)'
 
 
 class Processor:
@@ -64,6 +64,16 @@ def eval(self, args):
     def close(self, args):
         return args
 
+    def bytecode(self, args):
+        gremlin = args['gremlin']
+        args['gremlin'] = self._writer.toDict(gremlin)
+        aliases = args.get('aliases', '')
+        if not aliases:
+            aliases = {'g': 'g'}
+        args['aliases'] = aliases
+        return args
+
+
 class Traversal(Processor):
 
     def authentication(self, args):
@@ -187,6 +197,7 @@ def __init__(self, reader=None, writer=None, version=None):
         self._graphbinary_writer = writer
         self.standard = Standard(writer)
         self.traversal = Traversal(writer)
+        self.session = Session(writer)
 
     @property
     def version(self):
diff --git a/gremlin-python/src/main/python/gremlin_python/process/graph_traversal.py b/gremlin-python/src/main/python/gremlin_python/process/graph_traversal.py
index 698056f7650..3918be830e6 100644
--- a/gremlin-python/src/main/python/gremlin_python/process/graph_traversal.py
+++ b/gremlin-python/src/main/python/gremlin_python/process/graph_traversal.py
@@ -16,9 +16,10 @@
 # specific language governing permissions and limitations
 # under the License.
 #
-
+import logging
 import sys
 import copy
+from threading import Lock
 from .traversal import Traversal
 from .traversal import TraversalStrategies
 from .strategies import VertexProgramStrategy, OptionsStrategy
@@ -27,15 +28,19 @@
 from .. import statics
 from ..statics import long
 
+__author__ = 'Stephen Mallette (http://stephen.genoprime.com), Lyndon Bauto (lyndonb@bitquilltech.com)'
+
 
 class GraphTraversalSource(object):
     def __init__(self, graph, traversal_strategies, bytecode=None):
+        logging.info("Creating GraphTraversalSource.")
         self.graph = graph
         self.traversal_strategies = traversal_strategies
         if bytecode is None:
-          bytecode = Bytecode()
+            bytecode = Bytecode()
         self.bytecode = bytecode
         self.graph_traversal = GraphTraversal
+        self.remote_connection = None
 
     def __repr__(self):
         return "graphtraversalsource[" + str(self.graph) + "]"
@@ -79,7 +84,7 @@ def withoutStrategies(self, *args):
     def with_(self, k, v=None):
         source = self.get_graph_traversal_source()
         options_strategy = next((x for x in source.bytecode.source_instructions
-                                if x[0] == "withStrategies" and type(x[1]) is OptionsStrategy), None)
+                                 if x[0] == "withStrategies" and type(x[1]) is OptionsStrategy), None)
 
         val = True if v is None else v
         if options_strategy is None:
@@ -93,12 +98,31 @@ def with_(self, k, v=None):
     def withRemote(self, remote_connection):
         source = self.get_graph_traversal_source()
         source.traversal_strategies.add_strategies([RemoteStrategy(remote_connection)])
+        self.remote_connection = remote_connection
         return source
 
+    def tx(self):
+        # In order to keep the constructor unchanged within 3.5.x we can try to pop the RemoteConnection out of the
+        # TraversalStrategies. keeping this unchanged will allow user DSLs to not take a break.
+        # This is the same strategy as gremlin-javascript.
+        # TODO https://issues.apache.org/jira/browse/TINKERPOP-2664: refactor this to be nicer in 3.6.0 when
+        #  we can take a breaking change
+        remote_connection = next((x.remote_connection for x in self.traversal_strategies.traversal_strategies if
+                                  x.fqcn == "py:RemoteStrategy"), None)
+
+        if remote_connection is None:
+            raise Exception("Error, remote connection is required for transaction.")
+
+        # You can't do g.tx().begin().tx() i.e child transactions are not supported.
+        if remote_connection and remote_connection.is_session_bound():
+            raise Exception("This TraversalSource is already bound to a transaction - child transactions are not "
+                            "supported")
+        return Transaction(self, remote_connection)
+
     def withComputer(self, graph_computer=None, workers=None, result=None, persist=None, vertices=None,
                      edges=None, configuration=None):
-        return self.withStrategies(VertexProgramStrategy(graph_computer, workers, result, persist, vertices,
-                                   edges, configuration))
+        return self.withStrategies(
+            VertexProgramStrategy(graph_computer, workers, result, persist, vertices, edges, configuration))
 
     def E(self, *args):
         traversal = self.get_graph_traversal()
@@ -144,7 +168,7 @@ def __getitem__(self, index):
             if low == long(0):
                 return self.limit(high)
             else:
-                return self.range(low,high)
+                return self.range(low, high)
         else:
             raise TypeError("Index must be int or slice")
 
@@ -1003,6 +1027,69 @@ def sum(cls, *args):
         return cls.graph_traversal(None, None, Bytecode()).sum_(*args)
 
 
+# Class to handle transactions.
+class Transaction:
+
+    def __init__(self, g, remote_connection):
+        self._g = g
+        self._session_based_connection = None
+        self._remote_connection = remote_connection
+        self.__is_open = False
+        self.__mutex = Lock()
+
+    # Begins transaction.
+    def begin(self):
+        with self.__mutex:
+            # Verify transaction is not open.
+            self.__verify_transaction_state(False, "Transaction already started on this object")
+
+            # Create new session using the remote connection.
+            self._session_based_connection = self._remote_connection.create_session()
+            self.__is_open = True
+
+            # Set the session as a remote strategy within the traversal strategy.
+            traversal_strategy = TraversalStrategies()
+            traversal_strategy.add_strategies([RemoteStrategy(self._session_based_connection)])
+
+            # Return new GraphTraversalSource.
+            return GraphTraversalSource(self._g.graph, traversal_strategy, self._g.bytecode)
+
+    # Rolls transaction back.
+    def rollback(self):
+        with self.__mutex:
+            # Verify transaction is open, close session and return result of transaction's rollback.
+            self.__verify_transaction_state(True, "Cannot commit a transaction that is not started.")
+            return self.__close_session(self._session_based_connection.rollback())
+
+    # Commits the current transaction.
+    def commit(self):
+        with self.__mutex:
+            # Verify transaction is open, close session and return result of transaction's commit.
+            self.__verify_transaction_state(True, "Cannot commit a transaction that is not started.")
+            return self.__close_session(self._session_based_connection.commit())
+
+    # Closes session.
+    def close(self):
+        with self.__mutex:
+            # Verify transaction is open.
+            self.__verify_transaction_state(True, "Cannot close a transaction that has previously been closed.")
+            self.__close_session(None)
+
+    # Return whether or not transaction is open.
+    # Allow camelcase function here to keep api consistent with other languages.
+    def isOpen(self):
+        return self.__is_open
+
+    def __verify_transaction_state(self, state, error_message):
+        if self.__is_open != state:
+            raise Exception(error_message)
+
+    def __close_session(self, session):
+        self._session_based_connection.close()
+        self.__is_open = False
+        return session
+
+
 def V(*args):
     return __.V(*args)
 
@@ -1569,7 +1656,6 @@ def sum(*args):
 
 statics.add_static('where', where)
 
-
 # Deprecated - prefer the underscore suffixed versions e.g filter_()
 
 statics.add_static('filter', filter)
diff --git a/gremlin-python/src/main/python/gremlin_python/process/traversal.py b/gremlin-python/src/main/python/gremlin_python/process/traversal.py
index 0db602d01de..f7bd31d464a 100644
--- a/gremlin-python/src/main/python/gremlin_python/process/traversal.py
+++ b/gremlin-python/src/main/python/gremlin_python/process/traversal.py
@@ -538,7 +538,7 @@ class TraversalStrategies(object):
     global_cache = {}
 
     def __init__(self, traversal_strategies=None):
-        self.traversal_strategies =             traversal_strategies.traversal_strategies if traversal_strategies is not None else []
+        self.traversal_strategies = traversal_strategies.traversal_strategies if traversal_strategies is not None else []
 
     def add_strategies(self, traversal_strategies):
         self.traversal_strategies = self.traversal_strategies + traversal_strategies
@@ -654,6 +654,22 @@ def __repr__(self):
         return (str(self.source_instructions) if len(self.source_instructions) > 0 else "") + \
                (str(self.step_instructions) if len(self.step_instructions) > 0 else "")
 
+    @staticmethod
+    def _create_graph_op(name, *values):
+        bc = Bytecode()
+        bc.add_source(name, *values)
+        return bc
+
+    @staticmethod
+    class GraphOp:
+        @staticmethod
+        def commit():
+            return Bytecode._create_graph_op("tx", "commit")
+
+        @staticmethod
+        def rollback():
+            return Bytecode._create_graph_op("tx", "rollback")
+
 
 '''
 BINDINGS
diff --git a/gremlin-python/src/main/python/tests/conftest.py b/gremlin-python/src/main/python/tests/conftest.py
index d8582192139..fc3ef8bd9bf 100644
--- a/gremlin-python/src/main/python/tests/conftest.py
+++ b/gremlin-python/src/main/python/tests/conftest.py
@@ -20,8 +20,8 @@
 import concurrent.futures
 import ssl
 import pytest
-import sys
 import socket
+import logging
 
 from six.moves import queue
 
@@ -41,7 +41,10 @@
 basic_url = 'wss://localhost:{}/gremlin'.format(45941)
 kerberos_url = gremlin_server_url.format(45942)
 kerberized_service = 'test-service@{}'.format(socket.gethostname())
+verbose_logging = False
 
+logging.basicConfig(format='%(asctime)s [%(levelname)8s] [%(filename)15s:%(lineno)d - %(funcName)10s()] - %(message)s',
+                    level=logging.DEBUG if verbose_logging else logging.INFO)
 
 @pytest.fixture
 def connection(request):
@@ -60,6 +63,7 @@ def connection(request):
         def fin():
             executor.shutdown()
             conn.close()
+
         request.addfinalizer(fin)
         return conn
 
@@ -73,6 +77,7 @@ def client(request):
     else:
         def fin():
             client.close()
+
         request.addfinalizer(fin)
         return client
 
@@ -95,6 +100,7 @@ def authenticated_client(request):
     else:
         def fin():
             client.close()
+
         request.addfinalizer(fin)
         return client
 
@@ -118,6 +124,31 @@ def remote_connection(request):
     else:
         def fin():
             remote_conn.close()
+
+        request.addfinalizer(fin)
+        return remote_conn
+
+
+@pytest.fixture(params=['graphsonv2', 'graphsonv3', 'graphbinaryv1'])
+def remote_transaction_connection(request):
+    try:
+        if request.param == 'graphbinaryv1':
+            remote_conn = DriverRemoteConnection(anonymous_url, 'gtx',
+                                                 message_serializer=serializer.GraphBinarySerializersV1())
+        elif request.param == 'graphsonv2':
+            remote_conn = DriverRemoteConnection(anonymous_url, 'gtx',
+                                                 message_serializer=serializer.GraphSONSerializersV2d0())
+        elif request.param == 'graphsonv3':
+            remote_conn = DriverRemoteConnection(anonymous_url, 'gtx',
+                                                 message_serializer=serializer.GraphSONSerializersV3d0())
+        else:
+            raise ValueError("Invalid serializer option - " + request.param)
+    except OSError:
+        pytest.skip('Gremlin Server is not running')
+    else:
+        def fin():
+            remote_conn.close()
+
         request.addfinalizer(fin)
         return remote_conn
 
@@ -143,6 +174,7 @@ def remote_connection_authenticated(request):
     else:
         def fin():
             remote_conn.close()
+
         request.addfinalizer(fin)
         return remote_conn
 
@@ -157,6 +189,7 @@ def remote_connection_graphsonV2(request):
     else:
         def fin():
             remote_conn.close()
+
         request.addfinalizer(fin)
         return remote_conn
 
diff --git a/gremlin-python/src/main/python/tests/driver/test_client.py b/gremlin-python/src/main/python/tests/driver/test_client.py
index ac9664255e7..70694dcf278 100644
--- a/gremlin-python/src/main/python/tests/driver/test_client.py
+++ b/gremlin-python/src/main/python/tests/driver/test_client.py
@@ -169,7 +169,7 @@ def test_client_async(client):
     g = Graph().traversal()
     t = g.V()
     message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'gmodern'}})
-    future = client.submitAsync(message)
+    future = client.submit_async(message)
     result_set = future.result()
     assert len(result_set.all().result()) == 6
 
@@ -181,8 +181,8 @@ def test_connection_share(client):
     t = g.V()
     message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'gmodern'}})
     message2 = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'gmodern'}})
-    future = client.submitAsync(message)
-    future2 = client.submitAsync(message2)
+    future = client.submit_async(message)
+    future2 = client.submit_async(message2)
 
     result_set2 = future2.result()
     assert len(result_set2.all().result()) == 6
@@ -199,8 +199,8 @@ def test_multi_conn_pool(client):
     message = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'gmodern'}})
     message2 = RequestMessage('traversal', 'bytecode', {'gremlin': t.bytecode, 'aliases': {'g': 'gmodern'}})
     client = Client('ws://localhost:45940/gremlin', 'g', pool_size=1)
-    future = client.submitAsync(message)
-    future2 = client.submitAsync(message2)
+    future = client.submit_async(message)
+    future2 = client.submit_async(message2)
 
     result_set2 = future2.result()
     assert len(result_set2.all().result()) == 6
diff --git a/gremlin-python/src/main/python/tests/process/test_traversal.py b/gremlin-python/src/main/python/tests/process/test_traversal.py
index f731b843168..3df15dd53c7 100644
--- a/gremlin-python/src/main/python/tests/process/test_traversal.py
+++ b/gremlin-python/src/main/python/tests/process/test_traversal.py
@@ -17,8 +17,10 @@
 # under the License.
 #
 
-__author__ = 'Marko A. Rodriguez (http://markorodriguez.com)'
+__author__ = 'Marko A. Rodriguez (http://markorodriguez.com), Lyndon Bauto (lyndonb@bitquilltech.com)'
 
+import os
+import pytest
 from pytest import fail
 
 from gremlin_python.structure.graph import Graph
@@ -28,6 +30,10 @@
 from gremlin_python.process.graph_traversal import __
 
 
+def transactions_disabled():
+    return (os.environ['TEST_TRANSACTIONS'] != 'true') if 'TEST_TRANSACTIONS' in os.environ else False
+
+
 class TestTraversal(object):
     def test_bytecode(self):
         g = traversal().withGraph(Graph())
@@ -58,9 +64,9 @@ def test_bytecode(self):
         bytecode = g.V(Bindings.of('a', [1, 2, 3])) \
             .out(Bindings.of('b', 'created')) \
             .where(__.in_(Bindings.of('c', 'created'), Bindings.of('d', 'knows')) \
-            .count().is_(Bindings.of('e', P.gt(2)))).bytecode
+                   .count().is_(Bindings.of('e', P.gt(2)))).bytecode
         assert 5 == len(bytecode.bindings.keys())
-        assert [1,2,3] == bytecode.bindings['a']
+        assert [1, 2, 3] == bytecode.bindings['a']
         assert 'created' == bytecode.bindings['b']
         assert 'created' == bytecode.bindings['c']
         assert 'knows' == bytecode.bindings['d']
@@ -116,7 +122,8 @@ def test_no_sugar_for_magic_methods(self):
             t = g.V().__len__
             fail("can't do sugar with magic")
         except AttributeError as err:
-            assert str(err) == 'Python magic methods or keys starting with double underscore cannot be used for Gremlin sugar - prefer values(__len__)'
+            assert str(
+                err) == 'Python magic methods or keys starting with double underscore cannot be used for Gremlin sugar - prefer values(__len__)'
 
     def test_enforce_anonymous_child_traversal(self):
         g = traversal().withGraph(Graph())
@@ -124,8 +131,228 @@ def test_enforce_anonymous_child_traversal(self):
 
         try:
             g.V(0).addE("self").to(g.V(1))
-            assert false
+            assert False
         except TypeError:
             pass
 
+    @pytest.mark.skipif(transactions_disabled(), reason="Transactions are not enabled.")
+    def test_transaction_commit(self, remote_transaction_connection):
+        # Start a transaction traversal.
+        g = traversal().withRemote(remote_transaction_connection)
+        start_count = g.V().count().next()
+        tx = g.tx()
+
+        # Except transaction to not be open until begin is called.
+        assert not tx.isOpen()
+        gtx = tx.begin()
+        assert tx.isOpen()
+
+        add_node_validate_transaction_state(g, gtx, start_count, start_count, [tx])
+        add_node_validate_transaction_state(gtx, g, start_count + 2, start_count, [tx])
+
+        # Commit the transaction, this should close it and add our transaction only vertex to the graph,
+        # so our vertex count outside the transaction should be 2 + the start count.
+        tx.commit()
+        assert not tx.isOpen()
+        assert g.V().count().next() == start_count + 2
+
+        drop_graph_check_count(g)
+        verify_gtx_closed(gtx)
+
+    @pytest.mark.skipif(transactions_disabled(), reason="Transactions are not enabled.")
+    def test_transaction_rollback(self, remote_transaction_connection):
+        # Start a transaction traversal.
+        g = traversal().withRemote(remote_transaction_connection)
+        start_count = g.V().count().next()
+        tx = g.tx()
+
+        # Except transaction to not be open until begin is called.
+        assert not tx.isOpen()
+        gtx = tx.begin()
+        assert tx.isOpen()
+
+        add_node_validate_transaction_state(g, gtx, start_count, start_count, [tx])
+        add_node_validate_transaction_state(gtx, g, start_count + 2, start_count, [tx])
+
+        # Commit the transaction, this should close it and not add our transaction only vertex to the graph,
+        # so our vertex count outside the transaction should be the start count.
+        tx.rollback()
+        assert not tx.isOpen()
+        assert g.V().count().next() == start_count + 1
+
+        drop_graph_check_count(g)
+        verify_gtx_closed(gtx)
+
+    @pytest.mark.skipif(transactions_disabled(), reason="Transactions are not enabled.")
+    def test_transaction_no_begin(self, remote_transaction_connection):
+        # Start a transaction traversal.
+        g = traversal().withRemote(remote_transaction_connection)
+        tx = g.tx()
+
+        # Except transaction to not be open until begin is called.
+        assert not tx.isOpen()
+
+        try:
+            # Attempt to commit even though no tx is started.
+            tx().commit()
+            assert False
+        except Exception as e:
+            assert not tx.isOpen()
+
+        try:
+            # Attempt to rollback even though no tx is started.
+            tx().rollback()
+            assert False
+        except Exception as e:
+            assert not tx.isOpen()
+
+        try:
+            # Attempt to invoke tx().tx() which is illegal.
+            tx().tx()
+            assert False
+        except Exception as e:
+            assert not tx.isOpen()
+
+        # Create transaction and verify it is open.
+        gtx = tx.begin()
+        assert tx.isOpen()
+        try:
+            # Attempt to begin gtx which is illegal.
+            gtx().begin()
+            assert False
+        except Exception as e:
+            assert tx.isOpen()
+
+        # Commit this unused transaction and verify it is no longer open.
+        tx.commit()
+        assert not tx.isOpen()
+
+        # Create another transaction and verify it is open.
+        gtx = tx.begin()
+        assert tx.isOpen()
+
+        # Rollback this unused transaction and verify it is no longer open.
+        tx.rollback()
+        assert not tx.isOpen()
+
+    @pytest.mark.skipif(transactions_disabled(), reason="Transactions are not enabled.")
+    def test_multi_commit_transaction(self, remote_transaction_connection):
+        # Start a transaction traversal.
+        g = traversal().withRemote(remote_transaction_connection)
+        start_count = g.V().count().next()
+
+        # Create two transactions.
+        tx1, tx2 = g.tx(), g.tx()
+
+        # Generate two GraphTraversalSource's for each transaction with begin.
+        gtx1, gtx2 = tx1.begin(), tx2.begin()
+        verify_tx_state([tx1, tx2], True)
+
+        # Add node to gtx1, which should be visible to gtx1, not gtx2.
+        add_node_validate_transaction_state(g, gtx1, start_count, start_count, [tx1, tx2])
+
+        # Add node to gtx2, which should be visible to gtx2, not gtx2
+        add_node_validate_transaction_state(g, gtx2, start_count, start_count, [tx1, tx2])
+
+        # Add node to gtx1, which should be visible to gtx1, not gtx2. Note previous node also added.
+        add_node_validate_transaction_state(g, gtx1, start_count, start_count + 1, [tx1, tx2])
+
+        tx1.commit()
+        verify_tx_state([tx1], False)
+        verify_tx_state([tx2], True)
+        assert g.V().count().next() == start_count + 2
+
+        tx2.commit()
+        verify_tx_state([tx1, tx2], False)
+        assert g.V().count().next() == start_count + 3
+
+    @pytest.mark.skipif(transactions_disabled(), reason="Transactions are not enabled.")
+    def test_multi_rollback_transaction(self, remote_transaction_connection):
+        # Start a transaction traversal.
+        g = traversal().withRemote(remote_transaction_connection)
+        start_count = g.V().count().next()
+
+        # Create two transactions.
+        tx1, tx2 = g.tx(), g.tx()
+
+        # Generate two GraphTraversalSource's for each transaction with begin.
+        gtx1, gtx2 = tx1.begin(), tx2.begin()
+        verify_tx_state([tx1, tx2], True)
+
+        # Add node to gtx1, which should be visible to gtx1, not gtx2.
+        add_node_validate_transaction_state(g, gtx1, start_count, start_count, [tx1, tx2])
+
+        # Add node to gtx2, which should be visible to gtx2, not gtx2
+        add_node_validate_transaction_state(g, gtx2, start_count, start_count, [tx1, tx2])
+
+        # Add node to gtx1, which should be visible to gtx1, not gtx2. Note previous node also added.
+        add_node_validate_transaction_state(g, gtx1, start_count, start_count + 1, [tx1, tx2])
+
+        tx1.rollback()
+        verify_tx_state([tx1], False)
+        verify_tx_state([tx2], True)
+        assert g.V().count().next() == start_count
+
+        tx2.rollback()
+        verify_tx_state([tx1, tx2], False)
+        assert g.V().count().next() == start_count
+
+    @pytest.mark.skipif(transactions_disabled(), reason="Transactions are not enabled.")
+    def test_multi_commit_and_rollback(self, remote_transaction_connection):
+        # Start a transaction traversal.
+        g = traversal().withRemote(remote_transaction_connection)
+        start_count = g.V().count().next()
+
+        # Create two transactions.
+        tx1, tx2 = g.tx(), g.tx()
+
+        # Generate two GraphTraversalSource's for each transaction with begin.
+        gtx1, gtx2 = tx1.begin(), tx2.begin()
+        verify_tx_state([tx1, tx2], True)
+
+        # Add node to gtx1, which should be visible to gtx1, not gtx2.
+        add_node_validate_transaction_state(g, gtx1, start_count, start_count, [tx1, tx2])
+
+        # Add node to gtx2, which should be visible to gtx2, not gtx2
+        add_node_validate_transaction_state(g, gtx2, start_count, start_count, [tx1, tx2])
+
+        # Add node to gtx1, which should be visible to gtx1, not gtx2. Note previous node also added.
+        add_node_validate_transaction_state(g, gtx1, start_count, start_count + 1, [tx1, tx2])
+
+        tx1.commit()
+        verify_tx_state([tx1], False)
+        verify_tx_state([tx2], True)
+        assert g.V().count().next() == start_count + 2
+
+        tx2.rollback()
+        verify_tx_state([tx1, tx2], False)
+        assert g.V().count().next() == start_count + 2
+
+
+def add_node_validate_transaction_state(g, g_add_to, g_start_count, g_add_to_start_count, tx_verify_list):
+    # Add a single node to g_add_to, but not g.
+    # Check that vertex count in g is g_start_count and vertex count in g_add_to is g_add_to_start_count + 1.
+    g_add_to.addV("person").property("name", "lyndon").iterate()
+    assert g_add_to.V().count().next() == g_add_to_start_count + 1
+    assert g.V().count().next() == g_start_count
+    verify_tx_state(tx_verify_list, True)
+
+
+def verify_tx_state(gtx_list, value):
+    for tx in gtx_list:
+        assert tx.isOpen() == value
+
+
+def drop_graph_check_count(g):
+    g.V().drop().iterate()
+    assert g.V().count().next() == 0
+
 
+def verify_gtx_closed(gtx):
+    try:
+        # Attempt to add an additional vertex to the transaction. This should throw an exception since it
+        # has been rolled back.
+        gtx().addV("failure").iterate()
+        assert False
+    except Exception as e:
+        pass