set(Constant::TOPIC_CONF_REQUEST_REQUIRED_ACKS, $request_required_acks); } /** * @param mixed|string $acks * * producer需要server接收到数据之后发出的确认接收的信号,此项配置就是指procuder需要多少个这样的确认信号。此配置实际上代表了数据备份的可用性。以下设置为常用选项: * (1)acks=0: 设置为0表示producer不需要等待任何确认收到的信息。副本将立即加到socket buffer并认为已经发送。没有任何保障可以保证此种情况下server已经成功接收数据,同时重试配置不会发生作用(因为客户端不知道是否失败)回馈的offset会总是设置为-1; * (2)acks=1: 这意味着至少要等待leader已经成功将数据写入本地log,但是并没有等待所有follower是否成功写入。这种情况下,如果follower没有成功备份数据,而此时leader又挂掉,则消息会丢失。 * (3)acks=all: 这意味着leader需要等待所有备份都成功写入日志,这种策略会保证只要有一个备份存活就不会丢失数据。这是最强的保证。 * (4)其他的设置,例如acks=2也是可以的,这将需要给定的acks数量,但是这种策略一般很少用。 */ public function setAcks(int $acks): void { $this->set(Constant::TOPIC_CONF_ACKS, $acks); } /** * @param mixed|string $request_timeout_ms * * broker尽力实现request.required.acks需求时的等待时间,否则会发送错误到客户端 */ public function setRequestTimeoutMs(int $request_timeout_ms): void { $this->set(Constant::TOPIC_CONF_REQUEST_TIMEOUT_MS, $request_timeout_ms); } /** * @param mixed|string $message_timeout_ms * * 本地消息超时。此值仅在本地强制执行,并限制生成的消息等待成功传递的时间。0的时间是无限的。 * 这是librdkafka用于传递消息(包括重试)的最长时间。 * 超过重试计数或消息超时时发生传递错误。 * 如果配置了transactional.id,则消息超时将自动调整为transaction.timeout.ms。 */ public function setMessageTimeoutMs(int $message_timeout_ms): void { $this->set(Constant::TOPIC_CONF_MESSAGE_TIMEOUT_MS, $message_timeout_ms); } /** * @param mixed|string $delivery_timeout_ms * * Alias for message.timeout.ms: Local message timeout. * This value is only enforced locally and limits the time a produced message waits for successful delivery. * A time of 0 is infinite. * This is the maximum time librdkafka may use to deliver a message (including retries). * Delivery error occurs when either the retry count or the message timeout are exceeded. * The message timeout is automatically adjusted to transaction.timeout.ms if transactional.id is configured. */ public function setDeliveryTimeoutMs(int $delivery_timeout_ms): void { $this->set(Constant::TOPIC_CONF_DELIVERY_TIMEOUT_MS, $delivery_timeout_ms); } /** * @param mixed|string $queuing_strategy * * EXPERIMENTAL: subject to change or removal. * DEPRECATED Producer queuing strategy. * FIFO preserves produce ordering, while LIFO prioritizes new messages */ public function setQueuingStrategy(mixed $queuing_strategy): void { $this->set(Constant::TOPIC_CONF_QUEUING_STRATEGY, $queuing_strategy); } /** * @param mixed|string $produce_offset_report * * DEPRECATED No longer used. */ public function setProduceOffsetReport(bool $produce_offset_report): void { $this->set(Constant::TOPIC_CONF_PRODUCE_OFFSET_REPORT, $produce_offset_report); } /** * @param mixed|string $partitioner * * Partitioner: random - random distribution, * consistent - CRC32 hash of key (Empty and NULL keys are mapped to single partition), * consistent_random - CRC32 hash of key (Empty and NULL keys are randomly partitioned), * murmur2 - Java Producer compatible Murmur2 hash of key (NULL keys are mapped to single partition), * murmur2_random - Java Producer compatible Murmur2 hash of key (NULL keys are randomly partitioned. * This is functionally equivalent to the default partitioner in the Java Producer.), * fnv1a - FNV-1a hash of key (NULL keys are mapped to single partition), * fnv1a_random - FNV-1a hash of key (NULL keys are randomly partitioned). */ public function setPartitioner(mixed $partitioner): void { $this->set(Constant::TOPIC_CONF_PARTITIONER, $partitioner); } /** * @param mixed|string $partitioner_cb * * Custom partitioner callback (set with rd_kafka_topic_conf_set_partitioner_cb()) */ public function setPartitionerCb(mixed $partitioner_cb): void { $this->set(Constant::TOPIC_CONF_PARTITIONER_CB, $partitioner_cb); } /** * @param mixed|string $msg_order_cmp * * EXPERIMENTAL: subject to change or removal. * DEPRECATED Message queue ordering comparator (set with rd_kafka_topic_conf_set_msg_order_cmp()). * Also see queuing.strategy. * */ public function setMsgOrderCmp(mixed $msg_order_cmp): void { $this->set(Constant::TOPIC_CONF_MSG_ORDER_CMP, $msg_order_cmp); } /** * @param mixed|string $opaque * * Application opaque (set with rd_kafka_topic_conf_set_opaque()) */ public function setOpaque(mixed $opaque): void { $this->set(Constant::TOPIC_CONF_OPAQUE, $opaque); } /** * @param mixed|string $compression_codec * * Compression codec to use for compressing message sets. inherit = inherit global compression.codec configuration. */ public function setCompressionCodec(mixed $compression_codec): void { $this->set(Constant::TOPIC_CONF_COMPRESSION_CODEC, $compression_codec); } /** * @param mixed|string $compression_type * * Alias for compression.codec: compression codec to use for compressing message sets. * This is the default value for all topics, may be overridden by the topic configuration property compression.codec. */ public function setCompressionType(mixed $compression_type): void { $this->set(Constant::TOPIC_CONF_COMPRESSION_TYPE, $compression_type); } /** * @param mixed|string $compression_level * * Compression level parameter for algorithm selected by configuration property compression.codec. * Higher values will result in better compression at the cost of more CPU usage. * Usable range is algorithm-dependent: [0-9] for gzip; [0-12] for lz4; only 0 for snappy; * -1 = codec-dependent default compression level. */ public function setCompressionLevel(int $compression_level): void { $this->set(Constant::TOPIC_CONF_COMPRESSION_LEVEL, $compression_level); } /** * @param mixed|string $auto_commit_enable * * DEPRECATED [LEGACY PROPERTY: This property is used by the simple legacy consumer only. * When using the high-level KafkaConsumer, the global enable.auto.commit property must be used instead]. * If true, periodically commit offset of the last message handed to the application. * This committed offset will be used when the process restarts to pick up where it left off. * If false, the application will have to call rd_kafka_offset_store() to store an offset (optional). * Offsets will be written to broker or local file according to offset.store.method. */ public function setAutoCommitEnable(bool $auto_commit_enable): void { $this->set(Constant::TOPIC_CONF_AUTO_COMMIT_ENABLE, $auto_commit_enable); } /** * @param mixed|string $enable_auto_commit * * DEPRECATED Alias for auto.commit.enable: [LEGACY PROPERTY: This property is used by the simple legacy consumer only. * When using the high-level KafkaConsumer, the global enable.auto.commit property must be used instead]. * If true, periodically commit offset of the last message handed to the application. * This committed offset will be used when the process restarts to pick up where it left off. * If false, the application will have to call rd_kafka_offset_store() to store an offset (optional). * Offsets will be written to broker or local file according to offset.store.method. */ public function setEnableAutoCommit(bool $enable_auto_commit): void { $this->set(Constant::TOPIC_CONF_ENABLE_AUTO_COMMIT, $enable_auto_commit); } /** * @param mixed|string $auto_commit_interval_ms * * [LEGACY PROPERTY: This setting is used by the simple legacy consumer only. * When using the high-level KafkaConsumer, the global auto.commit.interval.ms property must be used instead]. * The frequency in milliseconds that the consumer offsets are committed (written) to offset storage. */ public function setAutoCommitIntervalMs(int $auto_commit_interval_ms): void { $this->set(Constant::TOPIC_CONF_AUTO_COMMIT_INTERVAL_MS, $auto_commit_interval_ms); } /** * @param mixed|string $auto_offset_reset * * Action to take when there is no initial offset in offset store or the desired offset is out of range: * 'smallest','earliest' - automatically reset the offset to the smallest offset, * 'largest','latest' - automatically reset the offset to the largest offset, * 'error' - trigger an error (ERR__AUTO_OFFSET_RESET) which is retrieved by consuming messages and checking 'message->err'. */ public function setAutoOffsetReset(mixed $auto_offset_reset): void { $this->set(Constant::TOPIC_CONF_AUTO_OFFSET_RESET, $auto_offset_reset); } /** * @param mixed|string $offset_store_path * * DEPRECATED Path to local file for storing offsets. * If the path is a directory a filename will be automatically generated in that directory based on the topic and partition. * File-based offset storage will be removed in a future version. */ public function setOffsetStorePath(string $offset_store_path): void { $this->set(Constant::TOPIC_CONF_OFFSET_STORE_PATH, $offset_store_path); } /** * @param mixed|string $offset_store_sync_interval_ms * * DEPRECATED fsync() interval for the offset file, in milliseconds. * Use -1 to disable syncing, and 0 for immediate sync after each write. * File-based offset storage will be removed in a future version. */ public function setOffsetStoreSyncIntervalMs(int $offset_store_sync_interval_ms): void { $this->set(Constant::TOPIC_CONF_OFFSET_STORE_SYNC_INTERVAL_MS, $offset_store_sync_interval_ms); } /** * @param mixed|string $offset_store_method * * DEPRECATED Offset commit store method: * 'file' - DEPRECATED: local file store (offset.store.path, et.al), * 'broker' - broker commit store (requires "group.id" to be configured and Apache Kafka 0.8.2 or later on the broker.). */ public function setOffsetStoreMethod(mixed $offset_store_method): void { $this->set(Constant::TOPIC_CONF_OFFSET_STORE_METHOD, $offset_store_method); } /** * @param mixed|string $consume_callback_max_messages * * Maximum number of messages to dispatch in one rd_kafka_consume_callback*() */ public function setConsumeCallbackMaxMessages(int $consume_callback_max_messages): void { $this->set(Constant::TOPIC_CONF_CONSUME_CALLBACK_MAX_MESSAGES, $consume_callback_max_messages); } }