Kafka源码解读——kafka配置文件_Configs

青旅半醒 2022-01-26 05:03 350阅读 0赞

## Kafka源码解读——配置文件\_Configs（间断持续更新......） ##

**目录**

*  Kafka源码解读——配置文件\_Configs（间断持续更新......）
 *  Producer Configs配置
 *  Consumer Configs配置
 *  Broker Configs配置
 *  Kafka Connect Configs配置

--------------------

以下配置中没有的请查看官网原文：[http://kafka.apache.org/documentation/][http_kafka.apache.org_documentation]

### Producer Configs配置 ###

<table> 
 <tbody> 
  <tr> 
   <td style="width:95px;"><strong>名称</strong></td> 
   <td><strong>释义</strong></td> 
  </tr> 
  <tr> 
   <td style="width:95px;">key.serializer&nbsp;&nbsp;</td> 
   <td> 
    <ul> 
     <li> <p>key序列化类</p> </li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:95px;">value.serializer&nbsp;&nbsp;</td> 
   <td> 
    <ul> 
     <li> <p>value序列化类</p> </li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:95px;">acks</td> 
   <td> 
    <ul> 
     <li> <p>设置为0，则生产者根本不会等待服务器的任何确认。记录将立即添加到套接字缓冲区并被视为已发送。在这种情况下，无法保证服务器已收到记录，重试配置将不会生效（因为客户端通常不知道任何故障）。为每个记录返回的偏移量将始终设置为-1</p> </li> 
     <li> <p>设置为1，这意味着领导者会将记录写入其本地日志，但不会等待所有追随者的完全确认。在这种情况下，如果领导者在确认记录后立即失败，但在追随者复制记录之前，则记录将丢失。&nbsp;</p> </li> 
     <li> <p>设置为all，这意味着领队将等待完整的同步副本集确认记录。这<span style="color:#ffbb66;">可以保证只要至少有一个同步副本保持活动状态，记录就不会丢失</span>。这是最有力的保证。这相当于ACKS=-1设置。&nbsp;</p> </li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:95px;">bootstrap.servers&nbsp;&nbsp;</td> 
   <td> 
    <ul> 
     <li> <p>&nbsp;用于建立与Kafka群集的初始连接的主机/端口对列表。客户机将使用所有服务器，而不管在此处指定用于引导的服务器是什么，此列表只影响用于发现完整服务器集的初始主机。此列表的格式应为host1:port1,host2:port2，…。由于这些服务器<span style="color:#ffbb66;">只是用于初始连接以发现完整的群集成员身份</span>（可能会动态更改），因此此列表不<span style="color:#ffbb66;">需要包含完整的服务器集</span>（但是，如果服务器关闭，您可能需要多个服务器）。&nbsp;</p> </li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:95px;">buffer.memory&nbsp;&nbsp; &nbsp;</td> 
   <td> 
    <ul> 
     <li>官方默认大小：33554432字节（32mb）</li> 
     <li>&nbsp;producer可用于缓冲等待发送到服务器的记录的内存总字节数。如果记录的发送速度超过了它们可以传递到服务器的速度，则生产者将阻塞，阻塞时间由ax.block.ms定义，之后将引发异常。&nbsp;</li> 
     <li>&nbsp;此设置应大致等于生产者将使用的总内存，但不是绝对的根据实际传输数据定义，因为并非生产者使用的所有内存都用于缓冲。一些额外的内存将用于压缩（如果启用了压缩）以及维护处理中的请求。&nbsp;</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:95px;">compression.type&nbsp;&nbsp;</td> 
   <td> 
    <ul> 
     <li>&nbsp;生产者生成的所有数据的压缩类型。默认值为无（即无压缩）。有效值为none、gzip、snappy、lz4或zstd。压缩是整批数据，因此分批的效果也会影响压缩比（越多分批意味着更好的压缩）。<img alt="" height="326" src="https://img-blog.csdnimg.cn/20190529180034390.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwZl85NDA4MTA2NTM4NDI=,size_16,color_FFFFFF,t_70" width="617"> 压缩参考：<a href="https://www.jianshu.com/p/d69e27749b00" rel="nofollow">https://www.jianshu.com/p/d69e27749b00</a></li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:95px;">retries</td> 
   <td> 
    <ul> 
     <li> <p>&nbsp;设置一个大于零的值将导致客户端重新发送任何发送失败并可能出现暂时性错误的记录。请注意，此重试与客户端在收到错误后重新发送记录的情况相同。如果不将max.in.flight.requests.per.connection设置为1，则允许重试可能会更改记录的顺序，因为如果将两个批发送到单个分区，而第一个批失败并重试，但第二个批成功，则第二批中的记录可能会首先出现。注意：另外，如果delivery.timeout.ms配置的超时在成功确认之前先过期，则在重试次数用完之前，生成请求将失败。通常，用户应该不设置此配置，而是使用delivery.timeout.ms来控制重试行为。&nbsp;</p> </li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:95px;">ssl.key.password&nbsp;&nbsp;</td> 
   <td> 
    <ul> 
     <li> <p>&nbsp;密钥存储文件中私钥的密码。这对于客户是可选的。&nbsp;</p> </li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:95px;">ssl.keystore.location&nbsp;&nbsp;</td> 
   <td> 
    <ul> 
     <li> <p>&nbsp;密钥存储文件的位置。这对于客户端是可选的，可以用于客户端的双向身份验证。&nbsp;</p> </li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:95px;">ssl.keystore.password&nbsp;&nbsp; &nbsp;</td> 
   <td> 
    <ul> 
     <li> <p>密钥存储文件的存储密码。这对于客户机是可选的，只有在ssl.keystore中才需要。位置配置。</p> </li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:95px;">ssl.truststore.location&nbsp;&nbsp; &nbsp;</td> 
   <td> 
    <ul> 
     <li> <p>信任存储文件的位置。</p> </li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:95px;">ssl.truststore.password&nbsp;&nbsp;</td> 
   <td> 
    <ul> 
     <li> <p>信任存储文件的密码。如果没有设置密码，对信任库的访问仍然可用，但已禁用完整性检查。</p> </li> 
    </ul></td> 
  </tr> 
 </tbody> 
</table>

### Consumer Configs配置 ###

<table> 
 <tbody> 
  <tr> 
   <td style="width:167px;"><strong>名称</strong></td> 
   <td style="width:682px;"><strong>释义</strong></td> 
  </tr> 
  <tr> 
   <td style="width:167px;">key.deserializer&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:682px;"> 
    <ul> 
     <li>同producer，不再解释</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:167px;">value.deserializer&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:682px;"> 
    <ul> 
     <li>同producer，不再解释</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:167px;">bootstrap.servers&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:682px;"> 
    <ul> 
     <li>同producer，不再解释</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:167px;">fetch.min.bytes&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:682px;"> 
    <ul> 
     <li>请求返回的最小数据量，如果请求的数据量不够这个值，则会继续请求，超过了请求时间之后，会立即fetch，将此值设置大于1可以提高请求吞吐量，缺点是会有延迟。</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:167px;">group.id&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:682px;"> 
    <ul> 
     <li>标识消费者组的唯一字符串，如果消费者使用subscribe(topic)或基于kafka的偏移管理策略，则需要配置该属性</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:167px;">heartbeat.interval.ms&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:682px;"> 
    <ul> 
     <li>用于确保消费者的会话保持活动的状态，或者在consumer加入、离开组时，保持组内平衡；</li> 
     <li>值必须低于<span style="color:#3399ea;">&lt;session.timeout.ms&gt;</span>，通常是不高于1/3，也可以更小，控制正常的、再平衡的预期时间</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:167px;">max.partition.fetch.bytes</td> 
   <td style="width:682px;"> 
    <ul> 
     <li>官方默认值：1048576字节（1mb）</li> 
     <li>服务器返给每个分区的最大数据量，由消费者批量获取，如果获取第一个非空分区的第一个批处理记录大于此值，则会回退，以防止consumer阻塞。</li> 
     <li>最大批记录大小由<span style="color:#3399ea;">&lt;message.max.bytes&gt;</span>（broker config）或<span style="color:#3399ea;">&lt;max.message.bytes&gt;</span>（topic config）</li> 
     <li>&nbsp;<span style="color:#3399ea;">&lt;fetch.max.bytes&gt;</span>用于限制consumer请求的记录大小</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:167px;">session.timeout.ms</td> 
   <td style="width:682px;"> 
    <ul> 
     <li>Kafka组管理者会检测consumer超时，consumer定期向broker节点发送心跳表示它还活着</li> 
     <li>如果在超时之前broker未收到consumer的心跳，那么该broker会移除当前consumer，consumer组会重新找一个consumer来消费</li> 
     <li>注意：此值的范围为：<span style="color:#3399ea;">&lt;group.min.session.timeout.ms&gt;</span>和<span style="color:#3399ea;">&lt;group.max.session.timeout.ms&gt;</span>之间（broker config）</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:167px;">ssl.key.password</td> 
   <td style="width:682px;"> 
    <ul> 
     <li>同producer，不再解释</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:167px;">ssl.keystore.location&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:682px;"> 
    <ul> 
     <li>同producer，不再解释</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:167px;">ssl.keystore.password&nbsp;</td> 
   <td style="width:682px;"> 
    <ul> 
     <li>同producer，不再解释</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:167px;">ssl.truststore.location&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:682px;"> 
    <ul> 
     <li>同producer，不再解释</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:167px;">ssl.truststore.password&nbsp;&nbsp;</td> 
   <td style="width:682px;"> 
    <ul> 
     <li>同producer，不再解释</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:167px;">auto.offset.reset&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:682px;"> 
    <ul> 
     <li>官方默认值：latest；取值范围 [latest, earliest, none]</li> 
     <li>如果在Kafka中没有初始化offet或者服务器上没有当前offset（例如：该数据已被删除该怎么办？）</li> 
    </ul> 
    <ol> 
     <li>earliest：自动将offset设置为最早的offset</li> 
     <li>latest：自动将offset设置为最新的offset</li> 
     <li>none：如果在consumer组里面没有找到之前的offset，则向该consumer抛异常</li> 
     <li>anything else：向consumer抛异常</li> 
    </ol></td> 
  </tr> 
  <tr> 
   <td style="width:167px;">&nbsp;</td> 
   <td style="width:682px;">&nbsp;</td> 
  </tr> 
  <tr> 
   <td style="width:167px;">&nbsp;</td> 
   <td style="width:682px;">&nbsp;</td> 
  </tr> 
  <tr> 
   <td style="width:167px;">&nbsp;</td> 
   <td style="width:682px;">&nbsp;</td> 
  </tr> 
 </tbody> 
</table>

### Broker Configs配置 ###

<table> 
 <tbody> 
  <tr> 
   <td style="width:161px;"><strong>名称</strong></td> 
   <td style="width:103px;">默认值</td> 
   <td style="width:585px;"><strong>说明</strong></td> 
  </tr> 
  <tr> 
   <td style="width:161px;">broker.id</td> 
   <td style="width:103px;">0</td> 
   <td style="width:585px;">每个broker的broker id都不能一样</td> 
  </tr> 
  <tr> 
   <td style="width:161px;">log.dirs</td> 
   <td style="width:103px;">/tmp/kafka-logs</td> 
   <td style="width:585px;">日志存放目录</td> 
  </tr> 
  <tr> 
   <td style="width:161px;">zookeeper.connect</td> 
   <td style="width:103px;">localhost:2181</td> 
   <td style="width:585px;">指定ZK连接：hostname:port，</td> 
  </tr> 
  <tr> 
   <td style="width:161px;">advertised.host.name</td> 
   <td style="width:103px;">&nbsp;</td> 
   <td style="width:585px;">&nbsp;</td> 
  </tr> 
  <tr> 
   <td style="width:161px;">&nbsp;</td> 
   <td style="width:103px;">&nbsp;</td> 
   <td style="width:585px;">&nbsp;</td> 
  </tr> 
  <tr> 
   <td style="width:161px;">&nbsp;</td> 
   <td style="width:103px;">&nbsp;</td> 
   <td style="width:585px;">&nbsp;</td> 
  </tr> 
  <tr> 
   <td style="width:161px;">&nbsp;</td> 
   <td style="width:103px;">&nbsp;</td> 
   <td style="width:585px;">&nbsp;</td> 
  </tr> 
  <tr> 
   <td style="width:161px;">&nbsp;</td> 
   <td style="width:103px;">&nbsp;</td> 
   <td style="width:585px;">&nbsp;</td> 
  </tr> 
  <tr> 
   <td style="width:161px;">&nbsp;</td> 
   <td style="width:103px;">&nbsp;</td> 
   <td style="width:585px;">&nbsp;</td> 
  </tr> 
  <tr> 
   <td style="width:161px;">&nbsp;</td> 
   <td style="width:103px;">&nbsp;</td> 
   <td style="width:585px;">&nbsp;</td> 
  </tr> 
  <tr> 
   <td style="width:161px;">&nbsp;</td> 
   <td style="width:103px;">&nbsp;</td> 
   <td style="width:585px;">&nbsp;</td> 
  </tr> 
  <tr> 
   <td style="width:161px;">&nbsp;</td> 
   <td style="width:103px;">&nbsp;</td> 
   <td style="width:585px;">&nbsp;</td> 
  </tr> 
  <tr> 
   <td style="width:161px;">&nbsp;</td> 
   <td style="width:103px;">&nbsp;</td> 
   <td style="width:585px;">&nbsp;</td> 
  </tr> 
  <tr> 
   <td style="width:161px;">&nbsp;</td> 
   <td style="width:103px;">&nbsp;</td> 
   <td style="width:585px;">&nbsp;</td> 
  </tr> 
  <tr> 
   <td style="width:161px;">&nbsp;</td> 
   <td style="width:103px;">&nbsp;</td> 
   <td style="width:585px;">&nbsp;</td> 
  </tr> 
  <tr> 
   <td style="width:161px;">&nbsp;</td> 
   <td style="width:103px;">&nbsp;</td> 
   <td style="width:585px;">&nbsp;</td> 
  </tr> 
 </tbody> 
</table>

###  Kafka Connect Configs配置 ###

<table> 
 <tbody> 
  <tr> 
   <td style="width:168px;"><strong>名称</strong></td> 
   <td style="width:681px;"><strong>释义</strong></td> 
  </tr> 
  <tr> 
   <td style="width:168px;">config.storage.topic&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:681px;"> 
    <ul> 
     <li>存储connector的topic name</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:168px;">group.id&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:681px;"> 
    <ul> 
     <li>标识woker连接的集群组的字符串</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:168px;">key.converter&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:681px;"> 
    <ul> 
     <li>转换器类，用于在Kafka连接格式和写入Kafka的序列化形式之间进行转换。控制了写或读Kafka的消息中的键的格式，因为它独立于connectors，所以它允许任何connectors使用任何序列化格式。常见的例子由Json、Avro。</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:168px;">offset.storage.topic&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:681px;"> 
    <ul> 
     <li>存储connectors偏移量的Kafka Topic的名称</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:168px;">status.storage.topic&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:681px;"> 
    <ul> 
     <li>存储connector和task状态的Kafka Topic名称</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:168px;">value.converter&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:681px;"> 
    <ul> 
     <li>与<span style="color:#3399ea;">&lt;key.converter&gt;</span>类似，这个设置value的转换器类。</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:168px;">bootstrap.servers&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:681px;"> 
    <ul> 
     <li>Kafka集群主机：端口列表</li> 
     <li>注意：不需要全部都配上，因为只是为了初始化连接，获取整个集群的数据，但为了保险起见，最好多配几个，防止部分主机宕机。</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:168px;">heartbeat.interval.ms&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:681px;"> 
    <ul> 
     <li>默认值：3000ms</li> 
     <li>用于确保worker的会话保持活动的状态，或者新的worker加入、离开组时，保持组内平衡；</li> 
     <li>值必须低于<span style="color:#3399ea;">&lt;session.timeout.ms&gt;</span>，通常是不高于1/3，也可以更小，控制正常的、再平衡的预期时间</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:168px;">rebalance.timeout.ms&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:681px;"> 
    <ul> 
     <li>默认值：60000ms</li> 
     <li>再平衡开始后，允许每个woker加入工作组的最大时间，基本上是对所有任务刷新数据量和提交偏移量所需的时间限制，超时之后，worker会从组内删除，导致提交offset失败。</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:168px;">session.timeout.ms&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:681px;"> 
    <ul> 
     <li>默认值：10000ms</li> 
     <li>用于检测worker故障导致的超时，worker定时发送心跳，表示自己还活着，如果该worker会话超时之前未收到心跳，那么worker将会从组内移除，并进行再平衡</li> 
     <li>注意：此值的范围为：&lt;group.min.session.timeout.ms&gt;和&lt;group.max.session.timeout.ms&gt;之间</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:168px;">ssl.key.password&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:681px;">&nbsp;</td> 
  </tr> 
  <tr> 
   <td style="width:168px;">ssl.keystore.location&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:681px;">&nbsp;</td> 
  </tr> 
  <tr> 
   <td style="width:168px;">ssl.keystore.password&nbsp;&nbsp;</td> 
   <td style="width:681px;">&nbsp;</td> 
  </tr> 
  <tr> 
   <td style="width:168px;">ssl.truststore.location&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:681px;">&nbsp;</td> 
  </tr> 
  <tr> 
   <td style="width:168px;">ssl.truststore.password&nbsp;&nbsp;</td> 
   <td style="width:681px;">&nbsp;</td> 
  </tr> 
  <tr> 
   <td style="width:168px;">client.dns.lookup&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:681px;"> 
    <ul> 
     <li>控制客户端如何使用DNS查找。</li> 
     <li>use_all_dns_ips：当查找返回一个主机的多个IP地址时，将在连接失败之前；用于引导和通知servers</li> 
     <li>resolve_canonical_bootstrap_servers_only：没一个都将被解析，单独作为一个规范名称列表</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:168px;">connections.max.idle.ms&nbsp;&nbsp;</td> 
   <td style="width:681px;"> 
    <ul> 
     <li>默认值：540000ms</li> 
     <li>配置多长时间之后关闭空闲连接</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:168px;">receive.buffer.bytes&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:681px;"> 
    <ul> 
     <li>读取数据时使用的TCP接收缓冲区(SO_RCVBUF)的大小，如果值为-1，则用系统默认值。</li> 
    </ul></td> 
  </tr> 
 </tbody> 
</table>

（1）Sources Connectors Configs配置

<table> 
 <tbody> 
  <tr> 
   <td style="width:217px;"><strong>名称</strong></td> 
   <td style="width:632px;"><strong>释义</strong></td> 
  </tr> 
  <tr> 
   <td style="width:217px;">name</td> 
   <td style="width:632px;"> 
    <ul> 
     <li>全局唯一的名称</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:217px;">connector.class&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:632px;"> 
    <ul> 
     <li>连接器的类别名，必须是org.apache.kafka.connect.connector.Connector的子类，如果是org.apache.kafka.connect.file.FileStreamSinkConnector，可以用全名，可以用FileStreamSink，可以用FileStreamSinkConnector，简写。</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:217px;">tasks.max&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:632px;"> 
    <ul> 
     <li>此连接器使用的最大任务数</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:217px;">errors.retry.timeout&nbsp;&nbsp;</td> 
   <td style="width:632px;"> 
    <ul> 
     <li>默认值：0</li> 
     <li>失败重连的最长持续时间，值为-1表示无限重试</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:217px;">errors.retry.delay.max.ms&nbsp;&nbsp;</td> 
   <td style="width:632px;"> 
    <ul> 
     <li>连续重试尝试之间的最长持续时间，一旦达到这个限制，抖动将增加到延迟，以防止出现群体问题（不太理解，估计就是怕影响到整个Kafka集群吧）</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:217px;">errors.tolerance&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:632px;"> 
    <ul> 
     <li>默认值：none</li> 
     <li>连接操作期间的容错行为，none表示任何错误都将导致连接任务失败，all表示跳过有问题的记录，</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:217px;">errors.log.enable&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:632px;"> 
    <ul> 
     <li>默认值：false</li> 
     <li>值为true，将每个错误、失败操作的详细信息、有问题的记录，都将写入连接程序的日志里。默认为false，因此只有不可容忍的错误才会报错。</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:217px;">errors.log.include.messages&nbsp;&nbsp;</td> 
   <td style="width:632px;"> 
    <ul> 
     <li>默认值：false</li> 
     <li>是否将导致失败的连接记录包含到日志中，默认是false，将组织记录key、value、header写入日志文件，尽管topic和分区号仍然会被log</li> 
    </ul></td> 
  </tr> 
 </tbody> 
</table>

（2）Sink Connectors Configs配置

<table> 
 <tbody> 
  <tr> 
   <td style="width:142px;"><strong>名称</strong></td> 
   <td><strong>释义</strong></td> 
  </tr> 
  <tr> 
   <td style="width:142px;">name</td> 
   <td> 
    <ul> 
     <li>全局唯一的标识</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:142px;">connector.class&nbsp;&nbsp; &nbsp;</td> 
   <td> 
    <ul> 
     <li>连接器类别名，必须是org.apache.kafka.connect.connector.Connector的子类，可以写全名、FileStreamSink、FileStreamSinkConnector，使配置更简单</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:142px;">tasks.max&nbsp;&nbsp; &nbsp;</td> 
   <td> 
    <ul> 
     <li>连接最大任务数</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:142px;">topics&nbsp;&nbsp; &nbsp;</td> 
   <td> 
    <ul> 
     <li>使用的topic列表，用','分隔</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:142px;">topics.regex&nbsp;&nbsp; &nbsp;</td> 
   <td> 
    <ul> 
     <li>topic列表的正则表达式，该正则会被编译成java.util.regex.Pattern，只能指定topic或topic.regex中的一个</li> 
    </ul></td> 
  </tr> 
 </tbody> 
</table>

四、Kafka Streams Configs配置

<table> 
 <tbody> 
  <tr> 
   <td style="width:196px;"><strong>名称</strong></td> 
   <td style="width:662px;"><strong>释义</strong></td> 
  </tr> 
  <tr> 
   <td style="width:196px;">application.id&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:662px;"> 
    <ul> 
     <li>集群内流程需唯一标识，用处有3个：</li> 
    </ul><p>1：默认客户端ID前缀</p> <p>2：成员管理组ID</p> <p>3：日志更新topic前缀</p> </td> 
  </tr> 
  <tr> 
   <td style="width:196px;">bootstrap.servers&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:662px;"> 
    <ul> 
     <li>略</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:196px;">replication.factor&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:662px;"> 
    <ul> 
     <li>默认值：1</li> 
     <li>流处理应用程序创建的更改日志topic和重分区topic的复制因子</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:196px;">state.dir&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:662px;"> 
    <ul> 
     <li>默认值：/tmp/kafka-streams&nbsp;&nbsp; &nbsp;</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:196px;">cache.max.bytes.buffering&nbsp; &nbsp;</td> 
   <td style="width:662px;"> 
    <ul> 
     <li>所有线程之间缓存的最大字节数</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td style="width:196px;">client.id&nbsp;&nbsp; &nbsp;</td> 
   <td style="width:662px;">用于内部的消费者、生产者和原始消费者的客户端ID的前缀，格式为“--streamthread--”</td> 
  </tr> 
 </tbody> 
</table>

[http_kafka.apache.org_documentation]: http://kafka.apache.org/documentation/