[table service] Fix python client can't read keys written by java clients

pom.xml

@@ -168,6 +168,7 @@
     <cobertura-maven-plugin.version>2.7</cobertura-maven-plugin.version>
     <coveralls-maven-plugin.version>4.3.0</coveralls-maven-plugin.version>
     <dockerfile-maven-plugin.version>1.3.7</dockerfile-maven-plugin.version>
+    <exec-maven-plugin.version>1.6.0</exec-maven-plugin.version>
     <license-maven-plugin.version>1.6</license-maven-plugin.version>
     <jacoco-maven-plugin.version>0.8.0</jacoco-maven-plugin.version>
     <maven-antrun-plugin.version>1.8</maven-antrun-plugin.version>

stream/clients/python/bookkeeper/admin/client.py

@@ -16,15 +16,13 @@
 
 import grpc
 import logging
-import pkg_resources
 
 from bookkeeper import types
 from bookkeeper.admin.namespace import Namespace
 from bookkeeper.admin.namespaces import Namespaces
 from bookkeeper.common.service_uri import ServiceURI
 from bookkeeper.proto.storage_pb2_grpc import RootRangeServiceStub
 
-__version__ = pkg_resources.get_distribution('bookkeeper').version
 __logger__ = logging.getLogger("bookkeeper.admin.Client")
 
 

stream/clients/python/bookkeeper/admin/namespace.py

@@ -14,9 +14,12 @@
 
 from __future__ import absolute_import
 
-from bookkeeper.common.constants import __DEFAULT_STREAM_CONF__
+from bookkeeper.common.constants import __DEFAULT_TABLE_CONF__
 from bookkeeper.common.constants import __ROOT_RANGE_METADATA__
 from bookkeeper.common.exceptions import from_root_range_rpc_response
+from bookkeeper.common.exceptions import InternalServerError
+from bookkeeper.common.exceptions import StreamExistsError
+from bookkeeper.common.exceptions import StreamNotFoundError
 from bookkeeper.common.method import wrap_method
 from bookkeeper.common.retry import Retry
 from bookkeeper.common.timeout import ExponentialTimeout
@@ -38,7 +41,7 @@ def __init__(self, root_range_service, namespace):
         self.__delete_with_retries__ =\
             wrap_method(self.__delete_stream__, self.__default_retry__)
 
-    def create(self, stream_name, stream_config=__DEFAULT_STREAM_CONF__):
+    def create(self, stream_name, stream_config=__DEFAULT_TABLE_CONF__):
         return self.__create_with_retries__(stream_name, stream_config)
 
     def __create_stream__(self, stream_name, stream_config):
@@ -51,8 +54,18 @@ def __create_stream__(self, stream_name, stream_config):
             request=create_stream_req,
             metadata=__ROOT_RANGE_METADATA__
         )
-        create_stream_resp = from_root_range_rpc_response(create_stream_resp)
-        return create_stream_resp.stream_props
+        try:
+            create_stream_resp = from_root_range_rpc_response(create_stream_resp)
+            return create_stream_resp.stream_props
+        except InternalServerError as ise:
+            # currently if a stream exists, it also throws
+            # internal server error
+            try:
+                self.get(stream_name=stream_name)
+                raise StreamExistsError("stream '%s' already exists at namespace '%s'"
+                                        % (stream_name, self.__namespace__))
+            except StreamNotFoundError:
+                raise ise
 
     def get(self, stream_name):
         return self.__get_with_retries__(stream_name)

stream/clients/python/bookkeeper/admin/namespaces.py

@@ -17,6 +17,9 @@
 from bookkeeper.common.constants import __DEFAULT_NS_CONF__
 from bookkeeper.common.constants import __ROOT_RANGE_METADATA__
 from bookkeeper.common.exceptions import from_root_range_rpc_response
+from bookkeeper.common.exceptions import InternalServerError
+from bookkeeper.common.exceptions import NamespaceExistsError
+from bookkeeper.common.exceptions import NamespaceNotFoundError
 from bookkeeper.common.method import wrap_method
 from bookkeeper.common.retry import Retry
 from bookkeeper.common.timeout import ExponentialTimeout
@@ -48,8 +51,17 @@ def __create_ns__(self, namespace, namespace_config):
             request=create_ns_req,
             metadata=__ROOT_RANGE_METADATA__
         )
-        create_ns_resp = from_root_range_rpc_response(create_ns_resp)
-        return create_ns_resp.ns_props
+        try:
+            create_ns_resp = from_root_range_rpc_response(create_ns_resp)
+            return create_ns_resp.ns_props
+        except InternalServerError as ise:
+            # currently if a namespace exists, it also throws
+            # internal server error.
+            try:
+                self.get(namespace=namespace)
+                raise NamespaceExistsError("namespace '%s' already exists" % namespace)
+            except NamespaceNotFoundError:
+                raise ise
 
     def get(self, namespace):
         return self.__get_with_retries__(namespace)

stream/clients/python/bookkeeper/common/constants.py

@@ -20,6 +20,28 @@
         ('bk-rt-sc-id-bin', util.to_bytes(__ROOT_RANGE_ID__, 8, "big"))
 ]
 __DEFAULT_STREAM_CONF__ = stream_pb2.StreamConfiguration(
+        key_type=stream_pb2.RangeKeyType.values()[0],
+        min_num_ranges=24,
+        initial_num_ranges=4,
+        split_policy=stream_pb2.SplitPolicy(
+                type=stream_pb2.SplitPolicyType.values()[0],
+                fixed_range_policy=stream_pb2.FixedRangeSplitPolicy(
+                        num_ranges=2
+                )
+        ),
+        rolling_policy=stream_pb2.SegmentRollingPolicy(
+                size_policy=stream_pb2.SizeBasedSegmentRollingPolicy(
+                        max_segment_size=128*1024*1024
+                )
+        ),
+        retention_policy=stream_pb2.RetentionPolicy(
+                time_policy=stream_pb2.TimeBasedRetentionPolicy(
+                        retention_minutes=-1
+                )
+        ),
+        storage_type=stream_pb2.StorageType.values()[0]
+)
+__DEFAULT_TABLE_CONF__ = stream_pb2.StreamConfiguration(
         key_type=stream_pb2.RangeKeyType.values()[0],
         min_num_ranges=24,
         initial_num_ranges=4,

stream/clients/python/bookkeeper/common/exceptions.py

@@ -98,7 +98,8 @@ def __init__(self, message, errors=(), response=None):
         self._response = response
 
     def __str__(self):
-        return '{} {}'.format(self.code, self.message)
+        return 'grpc_status_code = {}, bk_status_code = {} : {}'\
+            .format(self.grpc_status_code, self.bk_status_code, self.message)
 
     @property
     def errors(self):

stream/clients/python/bookkeeper/kv/client.py

@@ -16,15 +16,13 @@
 
 import grpc
 import logging
-import pkg_resources
 
 from bookkeeper import types
 from bookkeeper.admin.namespace import Namespace
 from bookkeeper.common.service_uri import ServiceURI
 from bookkeeper.kv.table import Table
 from bookkeeper.proto.storage_pb2_grpc import RootRangeServiceStub
 
-__version__ = pkg_resources.get_distribution('bookkeeper').version
 __logger__ = logging.getLogger("bookkeeper.kv.Client")
 
 

stream/clients/python/bookkeeper/kv/table.py

@@ -40,6 +40,8 @@ def __init__(self, channel, stream_props):
             wrap_method(self.__do_get__, self.__default_retry__)
         self.__del_with_retries__ =\
             wrap_method(self.__do_del__, self.__default_retry__)
+        self.__incr_with_retries__ =\
+            wrap_method(self.__do_incr__, self.__default_retry__)
         __logger__.info("initialized table instance with properties : %s",
                         stream_props)
 
@@ -79,6 +81,27 @@ def __do_put__(self, key, value, routing_header, grpc_metadata):
         )
         from_table_rpc_response(put_resp)
 
+    def incr_str(self, key_str, amount):
+        key = key_str.encode('utf-8')
+        return self.incr(key, amount)
+
+    def incr(self, key, amount):
+        metadata = self.__make_routing_metadata__(key)
+        header = self.__make_routing_header__(key)
+        return self.__incr_with_retries__(key, amount, header, metadata)
+
+    def __do_incr__(self, key, amount, routing_header, grpc_metadata):
+        incr_req = kv_rpc_pb2.IncrementRequest(
+            key=key,
+            amount=amount,
+            header=routing_header
+        )
+        incr_resp = self.__table_service__.Increment(
+            request=incr_req,
+            metadata=grpc_metadata
+        )
+        from_table_rpc_response(incr_resp)
+
     def get_str(self, key_str):
         key = key_str.encode('utf-8')
         return self.get(key)

stream/clients/python/nox.py

@@ -45,6 +45,33 @@ def default(session):
         *session.posargs
     )
 
+@nox.session
+def integration(session):
+    """Default integration test session.
+    This is intended to be run **without** an interpreter set, so
+    that the current ``python`` (on the ``PATH``) or the version of
+    Python corresponding to the ``nox`` binary the ``PATH`` can
+    run the tests.
+    """
+    # Install all test dependencies, then install local packages in-place.
+    session.install('pytest', 'pytest-cov')
+    for local_dep in LOCAL_DEPS:
+        session.install('-e', local_dep)
+    session.install('-e', '.')
+
+    # Run py.test against the unit tests.
+    session.run(
+        'py.test',
+        '--quiet',
+        '--cov-append',
+        '--cov-report=',
+        '--cov=bookkeeper',
+        '--cov-config=.coveragerc',
+        os.path.join('tests', 'integration'),
+        *session.posargs
+    )
+
+
 
 @nox.session
 def lint(session):

stream/clients/python/noxfile.py

@@ -52,6 +52,33 @@ def unit(session):
     default(session)
 
 
+@nox.session(python=[os.environ['PY_VERSION']])
+def integration(session):
+    """Default integration test session.
+    This is intended to be run **without** an interpreter set, so
+    that the current ``python`` (on the ``PATH``) or the version of
+    Python corresponding to the ``nox`` binary the ``PATH`` can
+    run the tests.
+    """
+    # Install all test dependencies, then install local packages in-place.
+    session.install('pytest', 'pytest-cov')
+    for local_dep in LOCAL_DEPS:
+        session.install('-e', local_dep)
+    session.install('-e', '.')
+
+    # Run py.test against the unit tests.
+    session.run(
+        'py.test',
+        '--quiet',
+        '--cov-append',
+        '--cov-report=',
+        '--cov=bookkeeper',
+        '--cov-config=.coveragerc',
+        os.path.join('tests', 'integration'),
+        *session.posargs
+    )
+
+
 @nox.session
 def lint(session):
     """Run linters.

stream/clients/python/scripts/docker_build.sh

@@ -0,0 +1,34 @@
+#!/bin/bash
+
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+set -e -x -u
+
+SCRIPT_DIR=$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )
+PY_VERSION=${PY_VERSION:-"3.7"}
+
+COMMANDS=`cat <<EOF
+pip install --upgrade setuptools wheel
+python3 setup.py sdist bdist_wheel
+EOF
+`
+
+docker run \
+    -v "${SCRIPT_DIR}/..":/opt/bookkeeper \
+    -w /opt/bookkeeper \
+    -e PY_VERSION=${PY_VERSION} \
+    python:${PY_VERSION} \
+    /bin/bash -c "${COMMANDS}"

stream/clients/python/scripts/docker_integration_tests.sh

@@ -0,0 +1,29 @@
+#!/bin/bash
+
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+set -e -x -u
+
+SCRIPT_DIR=$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )
+NOXSESSION=${NOXSESSION:-"integration"}
+
+docker run \
+    -v "${SCRIPT_DIR}/..":/opt/bookkeeper_python_client \
+    -w /opt/bookkeeper_python_client \
+    -e NOXSESSION="${NOXSESSION}" \
+    --entrypoint=/bin/bash \
+    apachebookkeeper/bookkeeper-current \
+    /opt/bookkeeper_python_client/scripts/run_integration_tests.sh

stream/clients/python/scripts/run_integration_tests.sh

@@ -0,0 +1,41 @@
+#!/bin/bash
+
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+set -e -x -u
+
+BK_HOME=/opt/bookkeeper
+
+echo "starting bookkeeper standalone ..."
+${BK_HOME}/bin/standalone process up
+
+echo "installing nox ..."
+find . | grep -E "(__pycache__|\.pyc|\.pyo$)" | xargs rm -rf
+pip install nox-automation
+echo "installed nox."
+
+TABLE="test-java-updates"
+echo "creating test table ..."
+${BK_HOME}/bin/bkctl tables create -r 1 ${TABLE}
+for x in {0..20}; do
+    echo "write kv pair '${x}'"
+    ${BK_HOME}/bin/bkctl table put ${TABLE} java-key-$x java-value-$x;
+done
+echo "ingested kv pairs for testing."
+
+echo "run integration tests"
+nox --session integration
+echo "done integration tests"

stream/clients/python/setup.py

@@ -19,7 +19,7 @@
 
 name = 'apache-bookkeeper-client'
 description = 'Apache BookKeeper client library'
-version = '4.9.0-alpha-0'
+version = '4.9.0-alpha-2'
 # Should be one of:
 # 'Development Status :: 3 - Alpha'
 # 'Development Status :: 4 - Beta'

stream/clients/python/tests/integration/bookkeeper/__init__.py

@@ -0,0 +1,11 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.