Add sphinx formatting to hyperlink methods (#898)

kafka/client.py

@@ -91,11 +91,11 @@ class SimpleClient(object):
         Returns the leader for a partition or None if the partition exists
         but has no leader.
 
-        UnknownTopicOrPartitionError will be raised if the topic or partition
+        Raises:
-        is not part of the metadata.
+            UnknownTopicOrPartitionError: If the topic or partition is not part
-
+                of the metadata.
-        LeaderNotAvailableError is raised if server has metadata, but there is
+            LeaderNotAvailableError: If the server has metadata, but there is no
-        no current leader
+        current leader.
         """
 
         key = TopicPartition(topic, partition)
@@ -434,8 +434,8 @@ class SimpleClient(object):
         Create an inactive copy of the client object, suitable for passing
         to a separate thread.
 
-        Note that the copied connections are not initialized, so reinit() must
+        Note that the copied connections are not initialized, so :meth:`.reinit`
-        be called on the returned copy.
+        must be called on the returned copy.
         """
         _conns = self._conns
         self._conns = {}

kafka/consumer/group.py

@@ -35,7 +35,7 @@ class KafkaConsumer(six.Iterator):
 
     Arguments:
         *topics (str): optional list of topics to subscribe to. If not set,
-            call subscribe() or assign() before consuming records.
+            call :meth:`.subscribe` or :meth:`.assign` before consuming records.
 
     Keyword Arguments:
         bootstrap_servers: 'host[:port]' string (or list of 'host[:port]'
@@ -119,7 +119,7 @@ class KafkaConsumer(six.Iterator):
         session_timeout_ms (int): The timeout used to detect failures when
             using Kafka's group management facilities. Default: 30000
         max_poll_records (int): The maximum number of records returned in a
-            single call to poll(). Default: 500
+            single call to :meth:`.poll`. Default: 500
         receive_buffer_bytes (int): The size of the TCP receive buffer
             (SO_RCVBUF) to use when reading data. Default: None (relies on
             system defaults). The java client defaults to 32768.
@@ -327,11 +327,11 @@ class KafkaConsumer(six.Iterator):
             partitions (list of TopicPartition): Assignment for this instance.
 
         Raises:
-            IllegalStateError: If consumer has already called subscribe()
+            IllegalStateError: If consumer has already called :meth:`.subscribe`.
 
         Warning:
             It is not possible to use both manual partition assignment with
-            assign() and group assignment with subscribe().
+            :meth:`.assign` and group assignment with :meth:`.subscribe`.
 
         Note:
             This interface does not support incremental assignment and will
@@ -349,12 +349,12 @@ class KafkaConsumer(six.Iterator):
     def assignment(self):
         """Get the TopicPartitions currently assigned to this consumer.
 
-        If partitions were directly assigned using assign(), then this will
+        If partitions were directly assigned using :meth:`.assign`, then this
-        simply return the same partitions that were previously assigned.
+        will simply return the same partitions that were previously assigned.
-        If topics were subscribed using subscribe(), then this will give the
+        If topics were subscribed using :meth:`.subscribe`, then this will give
-        set of topic partitions currently assigned to the consumer (which may
+        the set of topic partitions currently assigned to the consumer (which
-        be None if the assignment hasn't happened yet, or if the partitions are
+        may be None if the assignment hasn't happened yet, or if the partitions
-        in the process of being reassigned).
+        are in the process of being reassigned).
 
         Returns:
             set: {TopicPartition, ...}
@@ -518,7 +518,7 @@ class KafkaConsumer(six.Iterator):
                 with any records that are available currently in the buffer,
                 else returns empty. Must not be negative. Default: 0
             max_records (int, optional): The maximum number of records returned
-                in a single call to :meth:`poll`. Default: Inherit value from
+                in a single call to :meth:`.poll`. Default: Inherit value from
                 max_poll_records.
 
         Returns:
@@ -630,10 +630,10 @@ class KafkaConsumer(six.Iterator):
     def pause(self, *partitions):
         """Suspend fetching from the requested partitions.
 
-        Future calls to poll() will not return any records from these partitions
+        Future calls to :meth:`.poll` will not return any records from these
-        until they have been resumed using resume(). Note that this method does
+        partitions until they have been resumed using :meth:`.resume`. Note that
-        not affect partition subscription. In particular, it does not cause a
+        this method does not affect partition subscription. In particular, it
-        group rebalance when automatic assignment is used.
+        does not cause a group rebalance when automatic assignment is used.
 
         Arguments:
             *partitions (TopicPartition): Partitions to pause.
@@ -645,7 +645,7 @@ class KafkaConsumer(six.Iterator):
             self._subscription.pause(partition)
 
     def paused(self):
-        """Get the partitions that were previously paused by a call to pause().
+        """Get the partitions that were previously paused using :meth:`.pause`.
 
         Returns:
             set: {partition (TopicPartition), ...}
@@ -668,10 +668,10 @@ class KafkaConsumer(six.Iterator):
         """Manually specify the fetch offset for a TopicPartition.
 
         Overrides the fetch offsets that the consumer will use on the next
-        poll(). If this API is invoked for the same partition more than once,
+        :meth:`.poll`. If this API is invoked for the same partition more than
-        the latest offset will be used on the next poll(). Note that you may
+        once, the latest offset will be used on the next :meth:`.poll`. Note
-        lose data if this API is arbitrarily used in the middle of consumption,
+        that you may lose data if this API is arbitrarily used in the middle of
-        to reset the fetch offsets.
+        consumption, to reset the fetch offsets.
 
         Arguments:
             partition (TopicPartition): Partition for seek operation
@@ -743,7 +743,7 @@ class KafkaConsumer(six.Iterator):
         Topic subscriptions are not incremental: this list will replace the
         current assignment (if there is one).
 
-        This method is incompatible with assign().
+        This method is incompatible with :meth:`.assign`.
 
         Arguments:
             topics (list): List of topics for subscription.
@@ -772,7 +772,7 @@ class KafkaConsumer(six.Iterator):
                 through this interface are from topics subscribed in this call.
 
         Raises:
-            IllegalStateError: If called after previously calling assign().
+            IllegalStateError: If called after previously calling :meth:`.assign`.
             AssertionError: If neither topics or pattern is provided.
             TypeError: If listener is not a ConsumerRebalanceListener.
         """

kafka/coordinator/base.py

@@ -43,10 +43,10 @@ class BaseCoordinator(object):
        leader and begins processing.
 
     To leverage this protocol, an implementation must define the format of
-    metadata provided by each member for group registration in group_protocols()
+    metadata provided by each member for group registration in
-    and the format of the state assignment provided by the leader in
+    :meth:`.group_protocols` and the format of the state assignment provided by
-    _perform_assignment() and which becomes available to members in
+    the leader in :meth:`._perform_assignment` and which becomes available to
-    _on_join_complete().
+    members in :meth:`._on_join_complete`.
     """
 
     DEFAULT_CONFIG = {
@@ -277,7 +277,7 @@ class BaseCoordinator(object):
         """Join the group and return the assignment for the next generation.
 
         This function handles both JoinGroup and SyncGroup, delegating to
-        _perform_assignment() if elected leader by the coordinator.
+        :meth:`._perform_assignment` if elected leader by the coordinator.
 
         Returns:
             Future: resolves to the encoded-bytes assignment returned from the

kafka/producer/base.py

@@ -56,7 +56,8 @@ def _send_upstream(queue, client, codec, batch_time, batch_size,
     Messages placed on the queue should be tuples that conform to this format:
         ((topic, partition), message, key)
 
-    Currently does not mark messages with task_done. Do not attempt to join()!
+    Currently does not mark messages with task_done. Do not attempt to
+    :meth:`join`!
 
     Arguments:
         queue (threading.Queue): the queue from which to get messages
@@ -227,7 +228,8 @@ class Producer(object):
     Arguments:
         client (kafka.SimpleClient): instance to use for broker
             communications. If async=True, the background thread will use
-            client.copy(), which is expected to return a thread-safe object.
+            :meth:`client.copy`, which is expected to return a thread-safe
+            object.
         codec (kafka.protocol.ALL_CODECS): compression codec to use.
         req_acks (int, optional): A value indicating the acknowledgements that
             the server must receive before responding to the request,
@@ -263,7 +265,7 @@ class Producer(object):
             will not allow you to identify the specific message that failed,
             but it will allow you to match failures with retries.
         async_stop_timeout (int or float, optional): seconds to continue
-            attempting to send queued messages after producer.stop(),
+            attempting to send queued messages after :meth:`producer.stop`,
             defaults to 30.
 
     Deprecated Arguments:

kafka/producer/kafka.py

@@ -35,9 +35,9 @@ class KafkaProducer(object):
     thread that is responsible for turning these records into requests and
     transmitting them to the cluster.
 
-    The send() method is asynchronous. When called it adds the record to a
+    :meth:`.send` is asynchronous. When called it adds the record to a buffer of
-    buffer of pending record sends and immediately returns. This allows the
+    pending record sends and immediately returns. This allows the producer to
-    producer to batch together individual records for efficiency.
+    batch together individual records for efficiency.
 
     The 'acks' config controls the criteria under which requests are considered
     complete. The "all" setting will result in blocking on the full commit of
@@ -167,9 +167,9 @@ class KafkaProducer(object):
             will block up to max_block_ms, raising an exception on timeout.
             In the current implementation, this setting is an approximation.
             Default: 33554432 (32MB)
-        max_block_ms (int): Number of milliseconds to block during send() and
+        max_block_ms (int): Number of milliseconds to block during :meth:`.send`
-            partitions_for(). These methods can be blocked either because the
+            and :meth:`.partitions_for`. These methods can be blocked either
-            buffer is full or metadata unavailable. Blocking in the
+            because the buffer is full or metadata unavailable. Blocking in the
             user-supplied serializers or partitioner will not be counted against
             this timeout. Default: 60000.
         max_request_size (int): The maximum size of a request. This is also
@@ -537,8 +537,8 @@ class KafkaProducer(object):
         Invoking this method makes all buffered records immediately available
         to send (even if linger_ms is greater than 0) and blocks on the
         completion of the requests associated with these records. The
-        post-condition of flush() is that any previously sent record will have
+        post-condition of :meth:`.flush` is that any previously sent record will
-        completed (e.g. Future.is_done() == True). A request is considered
+        have completed (e.g. Future.is_done() == True). A request is considered
         completed when either it is successfully acknowledged according to the
         'acks' configuration for the producer, or it results in an error.