配置 kafka 同步刷盘

kafka.log.Log#append

// always update the last producer id map offset so that the snapshot reflects the current offset
// even if there isn‘t any idempotent data being written
producerStateManager.updateMapEndOffset(appendInfo.lastOffset + 1)

// increment the log end offset
updateLogEndOffset(appendInfo.lastOffset + 1)

// update the first unstable offset (which is used to compute LSO)
updateFirstUnstableOffset()

trace(s"Appended message set with last offset: ${appendInfo.lastOffset}, " +
  s"first offset: ${appendInfo.firstOffset}, " +
  s"next offset: ${nextOffsetMetadata.messageOffset}, " +
  s"and messages: $validRecords")

// 我们所需要的同步刷盘
if (unflushedMessages >= config.flushInterval)
  flush()

appendInfo

public static final String FLUSH_MESSAGES_INTERVAL_CONFIG = "flush.messages";
public static final String FLUSH_MESSAGES_INTERVAL_DOC = "This setting allows specifying an interval at " +
    "which we will force an fsync of data written to the log. For example if this was set to 1 " +
    "we would fsync after every message; if it were 5 we would fsync after every five messages. " +
    "In general we recommend you not set this and use replication for durability and allow the " +
    "operating system‘s background flush capabilities as it is more efficient. This setting can " +
    "be overridden on a per-topic basis (see <a href=\"#topicconfigs\">the per-topic configuration section</a>).";

public static final String FLUSH_MS_CONFIG = "flush.ms";
public static final String FLUSH_MS_DOC = "This setting allows specifying a time interval at which we will " +
    "force an fsync of data written to the log. For example if this was set to 1000 " +
    "we would fsync after 1000 ms had passed. In general we recommend you not set " +
    "this and use replication for durability and allow the operating system‘s background " +
    "flush capabilities as it is more efficient.";