Args: topics: A `tf.string` tensor containing topic names in [topic] format. For example: ["topic1", "topic2"] group_id: The id of the consumer group. For example: cgstream servers: An optional list of bootstrap servers. For example: `lo
(
self,
topics,
group_id,
servers,
stream_timeout=0,
message_poll_timeout=10000,
configuration=None,
internal=True,
)
| 686 | """ |
| 687 | |
| 688 | def __init__( |
| 689 | self, |
| 690 | topics, |
| 691 | group_id, |
| 692 | servers, |
| 693 | stream_timeout=0, |
| 694 | message_poll_timeout=10000, |
| 695 | configuration=None, |
| 696 | internal=True, |
| 697 | ): |
| 698 | """ |
| 699 | Args: |
| 700 | topics: A `tf.string` tensor containing topic names in [topic] format. |
| 701 | For example: ["topic1", "topic2"] |
| 702 | group_id: The id of the consumer group. For example: cgstream |
| 703 | servers: An optional list of bootstrap servers. |
| 704 | For example: `localhost:9092`. |
| 705 | stream_timeout: An optional timeout duration (in milliseconds) to block until |
| 706 | the new messages from kafka are fetched. |
| 707 | By default it is set to 0 milliseconds and doesn't block for new messages. |
| 708 | To block indefinitely, set it to -1. |
| 709 | message_poll_timeout: An optional timeout duration (in milliseconds) |
| 710 | after which the kafka consumer throws a timeout error while fetching |
| 711 | a single message. This value also represents the intervals at which |
| 712 | the kafka topic(s) are polled for new messages while using the `stream_timeout` |
| 713 | configuration: An optional `tf.string` tensor containing |
| 714 | configurations in [Key=Value] format. |
| 715 | Global configuration: please refer to 'Global configuration properties' |
| 716 | in librdkafka doc. Examples include |
| 717 | ["enable.auto.commit=false", "heartbeat.interval.ms=2000"] |
| 718 | Topic configuration: please refer to 'Topic configuration properties' |
| 719 | in librdkafka doc. Note all topic configurations should be |
| 720 | prefixed with `conf.topic.`. Examples include |
| 721 | ["conf.topic.auto.offset.reset=earliest"] |
| 722 | Reference: https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md |
| 723 | internal: Whether the dataset is being created from within the named scope. |
| 724 | Default: True |
| 725 | """ |
| 726 | with ops.name_scope("KafkaGroupIODataset"): |
| 727 | assert internal |
| 728 | |
| 729 | if stream_timeout == -1: |
| 730 | stream_timeout = sys.maxsize |
| 731 | elif stream_timeout >= 0: |
| 732 | # Taking the max of `stream_timeout` and `message_poll_timeout` |
| 733 | # to prevent the user from bothering about the underlying polling |
| 734 | # mechanism. |
| 735 | stream_timeout = max(stream_timeout, message_poll_timeout) |
| 736 | else: |
| 737 | raise ValueError( |
| 738 | "Invalid stream_timeout value: {} ,set it to -1 to block indefinitely." |
| 739 | .format(stream_timeout)) |
| 740 | metadata = list(configuration or []) |
| 741 | if group_id is not None: |
| 742 | metadata.append("group.id=%s" % group_id) |
| 743 | if servers is not None: |
| 744 | metadata.append("bootstrap.servers=%s" % servers) |
| 745 |