小编Mic*_*son的帖子

我打算跳过主题的开头，只读取从某个时间戳到结尾的消息。关于如何实现这一目标的任何提示？

我的主题有 3 个分区，我尝试使用以下代码从每个特定分区读取内容

from kafka import KafkaConsumer, TopicPartition

brokers = 'localhost:9092'
topic = 'b3'

m = KafkaConsumer(topic, bootstrap_servers=['localhost:9092'])
par = TopicPartition(topic=topic, partition=1)
m.assign(par)

但我收到此错误：

    raise IllegalStateError(self._SUBSCRIPTION_EXCEPTION_MESSAGE)
kafka.errors.IllegalStateError: IllegalStateError: You must choose only one way to configure your consumer: (1) subscribe to specific topics by name, (2) subscribe to topics matching a regex pattern, (3) assign itself specific topic-partitions.

有人可以帮我解决这个问题吗？

刚开始使用Spring Kafka（2.1.4.RELEASE）和Kafka（1.0.0）但是当我添加事务时，处理速度降低了很多。

代码：

spring.kafka.consumer.max-poll-records=10
spring.kafka.consumer.specific.avro.reader=true
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.group-id=${application.name}
spring.kafka.consumer.properties.isolation.level=read_committed
spring.kafka.consumer.key-deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer
spring.kafka.consumer.value-deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer

在 Java 中我添加了：

spring.kafka.consumer.max-poll-records=10
spring.kafka.consumer.specific.avro.reader=true
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.group-id=${application.name}
spring.kafka.consumer.properties.isolation.level=read_committed
spring.kafka.consumer.key-deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer
spring.kafka.consumer.value-deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer

当我删除该setTransactionManager(transactionManager)语句后，速度提高了很多。我做错了什么吗？

我正在尝试使用 ssl 创建 kafka 生产者。我需要有关如何在构造函数中设置 SSL 参数的信息，kafka-python 客户端中提供的信息不够描述。

ssl_certfile, ssl_cafile,ssl_keyfile参数是什么。我不确定在哪里可以找到这些文件。

producer = KafkaProducer(bootstrap_servers=kafka_broker,
  value_serializer=lambda v: json.dumps(v).encode('utf-8'),
  security_protocol='SSL',
  api_version=(0,10),
  ssl_cafile='ca-certs.pem',ssl_certfile='server.pem',
  ssl_keyfile='server.pem',ssl_password='xxx')
producer.send('rk976772_topic',{"test":0})

回溯（最近一次调用最后一次）：文件“”，第 1 行，在文件“/usr/lib/python2.7/site-packages/kafka/producer/kafka.py”中，第 543 行，在发送 self._wait_on_metadata(topic , self.config['max_block_ms'] / 1000.0) 文件 "/usr/lib/python2.7/site-packages/kafka/producer/kafka.py", line 664, in _wait_on_metadata "在 %.1f 后无法更新元数据秒。” % max_wait) kafka.errors.KafkaTimeoutError: KafkaTimeoutError: 60.0 秒后无法更新元数据。

我们在kafka属性文件中有broker.id属性。

Broker.id 的重要性是什么？如果我们在不同虚拟机上运行的所有 kafka 实例上为 Broker.id 分配相同的整数会怎样。

我目前开发了一个代码来显示主题、分区和日志偏移量。但我目前被困在如何获得分区的滞后上。我知道有一个 kafka offset 命令可以完成这个功能，但我需要的是一个 java 代码。

public static void main(String[] args) throws Exception {
    System.out.println("START CONSUMER");final Properties props = new Properties();
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUPID);
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

    // Create the consumer using props.
    final Consumer<Long, String> consumer =  new KafkaConsumer<>(props);

    // Subscribe to the topic.
    int i = 0;
    ArrayList<TopicPartition> partitions = new ArrayList<TopicPartition>();
    for (i=0;i<consumer.partitionsFor(TOPIC).size();i++)
    {
        TopicPartition partitiontemp = new TopicPartition(TOPIC, i);
        partitions.add(partitiontemp);
    }
    consumer.assign(partitions);
    consumer.seekToEnd(partitions);

    for (i=0;i<consumer.partitionsFor(TOPIC).size();i++)
    {
        System.out.printf("Topic: %s …

如果 kafkaenable.auto.commit=true 是否意味着每 5 秒它将在后台触发同步或异步操作以提交偏移量？

小间隔（2秒）会以某种方式影响延迟吗？

向 apache kafka 发布消息列表。任何人都可以使用 kafka api 提供示例代码，显示如何识别哪些消息已成功发布到主题，哪些消息从响应中失败？（请注意，我将在一个请求中批量发送消息列表。）

根据生产者配置，有：重试和max.in.flight.requests.per.connection。假设重试 > 0，最大飞行请求数/每次连接数 > 1。

消息是否可以在一个主题分区内乱序到达（例如，如果第一条消息已重试，但第二条消息却是第一次尝试传递给代理）？

还是仅在主题的多个分区之间发生乱序，但保留分区内的顺序？

我在使用bootstrap-server(即Kafka服务器)消费消息时遇到问题.知道为什么没有zookeeper就无法使用消息？

• 卡夫卡版本:kafka_2.11-1.0.0
• Zookeeper版本:kafka_2.11-1.0.0
• Zookeeper主机和端口:zkp02.mp.com:2181
• Kafka主机和港口:kfk03.mp.com:9092

制作一些消息:

[kfk03.mp.com ~]$ /bnsf/kafka/bin/kafka-console-producer.sh --broker-list kfk03.mp.com:9092 --topic test
>hi
>hi

如果我给出消费者,消费者无法使用消息–-bootstrap-server:

[kfk03.mp.com ~]$
/bnsf/kafka/bin/kafka-console-consumer.sh --bootstrap-server kfk03.mp.com:9092 --topic test --from-beginning

消费者能够在--zookeeper给出服务器而不是--bootstrap-server- 时使用消息:

[kfk03.mp.com ~]$ /bnsf/kafka/bin/kafka-console-consumer.sh --zookeeper zkp02.mp.com:2181 --topic test --from-beginning

Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper].

{"properties": {"messageType": "test", "sentDateTime": "2018-02-25T21:46:00.000+0000"}, "name": "Uttam Anand", "age": 29}
{"properties": {"messageType": …

我们无法理解 DMARC 报告。下面是我们从雅虎那里得到的。

DMARC 报告

<?xml version="1.0"?>   
<feedback>  
  <report_metadata> 
    <org_name>Yahoo! Inc.</org_name>    
    <email>postmaster@dmarc.yahoo.com</email>   
    <report_id>1519264633.286724</report_id>    
    <date_range>    
      <begin>1519171200</begin> 
      <end>1519257599 </end>    
    </date_range>   
  </report_metadata>    
  <policy_published>    
    <domain>mydomain.com</domain>   
    <adkim>r</adkim>    
    <aspf>r</aspf>  
    <p>none</p> 
    <pct>100</pct>  
  </policy_published>   
  <record>  
    <row>   
      <source_ip>198.210.47.11</source_ip>  
      <count>12</count> 
      <policy_evaluated>    
        <disposition>none</disposition> 
        <dkim>fail</dkim>   
        <spf>fail</spf> 
      </policy_evaluated>   
    </row>  
    <identifiers>   
      <header_from>mydomain.com</header_from>   
    </identifiers>  
    <auth_results>  
      <dkim>    
        <domain>woodersmtp.com</domain> 
        <result>pass</result>   
      </dkim>   
      <spf> 
        <domain>woodersmtp.com</domain> 
        <result>pass</result>   
      </spf>    
    </auth_results> 
  </record>

在这份报告中，SPF 和 DKIM 失败但同样通过。也没有得到，这个“woodersmtp.com”是什么？这不是我的域。

总的来说，我想知道：

1. 这些“woodersmtp.com”是什么，为什么来自我域的电子邮件会通过它？

2. 如何确保 SPF 和 DKIM 仅通过合法电子邮件？

3. 如何找出所有这些失败电子邮件的消息标题？