THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
...
Please see the details in KIP-33.
Option discussions with use cases
This section discusses how the three options work with a few use cases. Option 1 and Option 2 are in the rejected option section.
Options comparison
| Use cases | Option 1 (Message contains CreateTime + LogAppendTime) | option 2 (Message contains LogAppendTime only) | option 3 (message contains CreateTime only, brokers keep LogAppendTime in log index) | option 4 (Message contains a timestamp that could be overridden by broker depending on configured time difference threshold) | Comparison |
|---|---|---|---|---|---|
| Mirror Maker | Broker overrides the LAT and keep the CT as is. | Broker overrides the LAT | Broker keep the CT as. And add index entry with LAT to the log index file. | Mirror maker will keep the consumed messages' timestamp. Those timestamp may or may not be overwritten by target cluster depending on the configuration. | Option 1 provides the most information to user. The only concern is whether we should expose LAT to user. Option 2 loses the CreateTime information. Option 3 have same amount information as option 1 from broker point of view. From user point of view, it does not expose the LAT. Option 4 could be equivalent to either having CreateTime only or having LogAppendTime only depending on configuration. |
| Log Retention | Broker will use the LAT of the last message in a segment to enforce the policy. | Same as option 1. | Broker will use the LAT of the last entry in the log index file to enforce the retention policy. Because the leader is the source of truth for LAT, followers need to get the LAT from leader when they replicate the messages. That means we need to introduce a new wire protocol to fetch the time based log index file as well. When log recovery happens, the rebuilt time index would have different LAT from the actual arrival time of the messages in the log. And the LAT in the index file will be very close, or even the same. | The broker will take a look at the last Time Index entry of the segment to decide whether to delete the log segment or not. | Option 1 and option 2 can work with existing replication design and solve the log retention issue we have now. Option 3 solves the issue but is a bit involved because it needs to replicate the time index entry as well. Option 4 could be equivalent to either having CreateTime only or having LogAppendTime only depending on configuration. |
| Log rolling | Broker will use the LAT of the first message in a log segment to enforce the policy. | Same as option 1. | Broker will use the LAT of the first entry in the log index file to enforce the retention policy. Similar to the log retention case, the followers needs to replicate the time index as well. The log recovery happens, the log rolling might not be honored either. | The broker will keep an in memory earliest message timestamp in the active log segment. On broker startup, the broker will need to scan the messages in the active segment to find the earliest timestamp. | Option 1 and option 2 solves the log rolling issue. Option 3 solves the issue but is a bit involved because it needs to replicate the time index entry as well. On broker start up, Option 4 needs to scan the active log segment to find the earliest timestamp in the log segment. |
| Stream processing | Applications don't need to include the CreateTime in the payload but simply use the CreateTime field. | Applications have to put CreateTime into the payload. | Applications don't need to include the CreateTime in the payload but simply use the CreateTime field. | Option 4 could be equivalent to either having CreateTime only or having LogAppendTime only depending on configuration. | The benefit of having a CreateTime with each message rather than put it into payload is that application protocol can be simplified. It is convenient for the infrastructure to provide the timestamp so there is no need for each application to worry about the timestamp. |
| Latency measurement | User can get End2End latency and lag in time. | User can get the lag in time. | User can get End2End latency. | Depending on the max.message.time.difference.ms configuration. User may or may not be able to find out the latency. | Option 1 has most information for user. |
| Search message by timestamp. | Detail discussion in KIP-33 | Detail discussion in KIP-33 | Detail discussion in KIP-33 | Detail discussion in KIP-33 | Detail discussion in KIP-33 |
Mirror maker case in detail
The behavior of broker for all the options are the same: The broker will always override the LogAppendTime(if exists) when message arrives the broker and keep the CreateTime(if exists) untouched.
The broker does not distinguish mirror maker from other producers. The following example explains what will the timestamp look like when there is mirror maker in the picture.(CT - CreateTime, LAT - LogAppendTime)
- Application producer produces message at T0. ( [CT = T0, LAT = -1] )
- The broker in cluster1 received message at T1 = T0 + latency1 and appended the message to the log. Latency1 includes linger.ms and some other latency. ( [CT = T0, LAT = T1] )
- Mirror maker copies the message to broker in cluster2. ( [CT = T0, LAT = T1] )
- The broker in cluster2 receives message at T2 = T1 + latency2 and appended the message to the log. ( [CT = T0, LAT = T2] )
The CreateTime of a message in source cluster and target cluster will be same. i.e. the timestamp is passed across clusters.
The LogAppendTime of a message in source cluster and target cluster will be different.
Discussion: should we use CreateTime OR LogAppendTime for log retention and time based log rolling?
To discuss the usage of CreateTime and LogAppendTime, it is useful to summarize the latency pattern of the messages flowing through the pipeline. The latency can be summarized to the following pattern:
- the messages flow through the pipeline with small latency.
- the messages flow through the pipeline with similar large latency.
- the messages flow through the pipeline with large latency difference.
Also it would be useful to think about the impact of a completely wrong timestamp set by client. i.e. the robustness of the system.
Log Retention
There are both pros and cons for log retention to be based on CreateTime or LogAppendTime.
- Pattern 1:
Because the latency is small, so CreateTime and LogAppendTime will be close and their won't be much difference. - pattern 2:
If the log retention is based on the message creation time, it will not be affected by the latency in the data pipeline because the send time will not change.
If the log retention is based on the LogAppendTime, it will be affected by the latency in the pipeline. Because of the latency difference, some data can be deleted on one cluster, but not on another cluster in the pipeline. - Pattern 3:
When the messages with significantly different timestamp goes into a cluster at around same time, the retention policy is hard to follow if we use CreateTime. For example, imagine two mirror makers copy data from two source clusters to the same target cluster. If MirrorMaker1 is copying Messages with CreateTime around 1:00 PM today, and MirrorMaker2 is copying messages with CreateTime around 1:00 PM yesterday. Those messages can go to the same log segment in the target cluster. It will be difficult for broker to apply retention policy to the log segment. The broker needs to maintain the knowledge about the latest CreateTime of all messages in a log segment and persist the information somewhere. - Robustness:
If there is a message with CreateTime set to the future, the log might be kept for very long. Broker needs to sanity check the timestamp when receive the message. It could by tricky to determine which timestamp is not valid
Comparison:
| pattern 1 | pattern 2 | pattern 3 | Robustness | |
|---|---|---|---|---|
| Preference | CT or LAT | CT | LAT | LAT |
In reality, we usually don't see all the pipeline has same large latency, so it looks LogAppendTime is preferable than CreateTime for log retention.
Time based log rolling
The main purpose of time based log rolling is to avoid the situation where a low volume topic always has only one segment which is also the active segment. From its nature, server side log rolling makes more sense.
- Pattern 1:
Because the latency is small, so CreateTime and LogAppendTime will be close and their won't be much difference. - Pattern 2:
When the latency is large, it is possible that when a new message is produced to a broker, the CreateTime has already reached the rolling criteria. This might cause a segment with only one message. - Pattern 3:
Similar to pattern 2, a lagged message might result in a single message log segment. - Robustness:
Similar to as pattern 2 and pattern 3. Also a CreateTime in the future might break the log rolling as well.
| pattern 1 | pattern 2 | pattern 3 | Robustness | |
|---|---|---|---|---|
| Preference | CT or LAT | LAT | LAT | LAT |
Compatibility, Deprecation, and Migration Plan
NOTE: This part is drafted based on the assumption that KIP-31 and KIP-32 will be implemented in one patch.
The proposed protocol is not backward compatible. The migration plan are as below:
Phase 1 (MessageAndOffset V0 on disk):
- Set message.format.version=0 on brokers. (Broker will write MessageAndOffset V0 to disk)
- Create internal ApiVersion 0.9.0-1** which uses ProducerRequest V2 and FetchRequest V2.
- Configure the broker to use ApiVersion 0.9.0 (ProduceRequest V1 and FetchRequest V1).
- Do a rolling upgrade of the brokers to let the broker pick up the new code supporting ApiVersion 0.9.0-1.
- Bump up ApiVersion of broker to 0.9.0-1 to let the broker use FetchRequest V2 for replication.
- Upgraded brokers support both ProducerRequest V2 and FetchRequest V2 which uses magic byte 1 for MessageAndOffset.
- When broker sees a producer request V1 (MessageAndOffset = V0), it will decompress the message, assign offsets using absolute offsets and re-compress the message.
- When broker sees a producer request V2 (MessageAndOffset = V1), it will decompress the message, assign offsets using absolute offsets, ignore the time field and do re-compression. i.e. down-convert the message format to MessageAndOffset V0.
- When broker sees a fetch request V1 (Supporting MessageAndOffset = V0), because the data format on disk is MessageAndOffset V0, it will use the zero-copy transfer to reply with fetch response V1 with MessageAndOffset V0.
- When broker sees a fetch request V2 (Supporting MessageAndOffset = V0, V1), because the data format on disk is MessageAndOffset V0, it will use zero-copy transfer to reply with fetch response V2 with MessageAndOffset V0.
- When broker sees a producer request V1 (MessageAndOffset = V0), it will decompress the message, assign offsets using absolute offsets and re-compress the message.
- Upgrade consumer to send FetchRequest V2.
- Upgrade producer to send ProducerRequest V2.
Phase 2 (MessageAndOffset V1 on disk):
- After most of the consumers are upgraded, Bump up message.format.version=1 and rolling bounce the brokers.
- Upgraded brokers do the followings:
- When broker sees a producer request V1 (MessageAndOffset = V0), it will decompress the message, assign offsets using relative offsets, fill in the time field with current server time and re-compress the message. i.e. up-convert the message format to MessageAndOffset V1.
- When broker sees a producer request V2 (MessageAndOffset = V1), it will decompress the message, assign offsets using relative offsets, check and maybe overwrite the time field, and NOT do re-compression.
- When broker sees a fetch request V1 (Supporting MessageAndOffset = V0), because the data format on disk is MessageAndOffset V1, it will NOT use the zero-copy transfer. Instead the broker will read the message from disk, down-convert them to V0 and reply using fetch response V1 with MessageAndOffset V0.
- When broker sees a fetch request V2 (Supporting MessageAndOffset = V0, V1), because the data format on disk is MessageAndOffset V1, it will use zero-copy transfer to reply with fetch response V2 with MessageAndOffset V1.
- When broker sees a producer request V1 (MessageAndOffset = V0), it will decompress the message, assign offsets using relative offsets, fill in the time field with current server time and re-compress the message. i.e. up-convert the message format to MessageAndOffset V1.
For producer, there will be no impact.
In phase 1, there will be no impact for consumers.
In phase 2, there will be some performance penalty for consumers that only supports MessageAndOffset V0, because there is no zero-copy transfer.
At the beginning of phase 2, there will be some time the log segment contains both MessageAndOffset V0 and V1. The broker will always do down conversion for FetchRequest V1 and zero-copy transfer for FetchRequest V2.
** We introduce internal ApiVersion here to help the user who are running on trunk to upgrade in the future. Otherwise the interim ApiVersion between two official releases will require users to downgrade ApiVersion then upgrade.
To canary a broker
After phase 1, it is possible for user to canary a broker in phase 2 and roll back if something goes wrong. The procedure is:
- Set message.format.version=1 on one of the brokers (broker B).
- Broker B will start to act like what described in phase 2.
- It will sends FetchRequest V2 to other brokers for replication.
- It will only see ProduceRequest/FetchRequest V1 from other brokers and clietns.
- If something goes wrong, we can do the following to rollback:
- shutdown broker B
- nuke the data of the topics it was serving as leader before shutdown
- set message.format.version=0
- restart the broker to let the broker replicate from leaders. At this point the data on disk will be in MessageAndOffset V0.
- shutdown broker B
In step 2, it is recommended to put only small amount of leaders on the broker, because at that point the broker needs to do down conversion for all the fetch requests.
...
Compatibility, Deprecation, and Migration Plan
NOTE: This part is drafted based on the assumption that KIP-31 and KIP-32 will be implemented in one patch.
The proposed protocol is not backward compatible. The migration plan are as below:
Phase 1 (MessageAndOffset V0 on disk):
- Set message.format.version=0 on brokers. (Broker will write MessageAndOffset V0 to disk)
- Create internal ApiVersion 0.9.0-1** which uses ProducerRequest V2 and FetchRequest V2.
- Configure the broker to use ApiVersion 0.9.0 (ProduceRequest V1 and FetchRequest V1).
- Do a rolling upgrade of the brokers to let the broker pick up the new code supporting ApiVersion 0.9.0-1.
- Bump up ApiVersion of broker to 0.9.0-1 to let the broker use FetchRequest V2 for replication.
- Upgraded brokers support both ProducerRequest V2 and FetchRequest V2 which uses magic byte 1 for MessageAndOffset.
- When broker sees a producer request V1 (MessageAndOffset = V0), it will decompress the message, assign offsets using absolute offsets and re-compress the message.
- When broker sees a producer request V2 (MessageAndOffset = V1), it will decompress the message, assign offsets using absolute offsets, ignore the time field and do re-compression. i.e. down-convert the message format to MessageAndOffset V0.
- When broker sees a fetch request V1 (Supporting MessageAndOffset = V0), because the data format on disk is MessageAndOffset V0, it will use the zero-copy transfer to reply with fetch response V1 with MessageAndOffset V0.
- When broker sees a fetch request V2 (Supporting MessageAndOffset = V0, V1), because the data format on disk is MessageAndOffset V0, it will use zero-copy transfer to reply with fetch response V2 with MessageAndOffset V0.
- When broker sees a producer request V1 (MessageAndOffset = V0), it will decompress the message, assign offsets using absolute offsets and re-compress the message.
- Upgrade consumer to send FetchRequest V2.
- Upgrade producer to send ProducerRequest V2.
Phase 2 (MessageAndOffset V1 on disk):
- After most of the consumers are upgraded, Bump up message.format.version=1 and rolling bounce the brokers.
- Upgraded brokers do the followings:
- When broker sees a producer request V1 (MessageAndOffset = V0), it will decompress the message, assign offsets using relative offsets, fill in the time field with current server time and re-compress the message. i.e. up-convert the message format to MessageAndOffset V1.
- When broker sees a producer request V2 (MessageAndOffset = V1), it will decompress the message, assign offsets using relative offsets, check and maybe overwrite the time field, and NOT do re-compression.
- When broker sees a fetch request V1 (Supporting MessageAndOffset = V0), because the data format on disk is MessageAndOffset V1, it will NOT use the zero-copy transfer. Instead the broker will read the message from disk, down-convert them to V0 and reply using fetch response V1 with MessageAndOffset V0.
- When broker sees a fetch request V2 (Supporting MessageAndOffset = V0, V1), because the data format on disk is MessageAndOffset V1, it will use zero-copy transfer to reply with fetch response V2 with MessageAndOffset V1.
- When broker sees a producer request V1 (MessageAndOffset = V0), it will decompress the message, assign offsets using relative offsets, fill in the time field with current server time and re-compress the message. i.e. up-convert the message format to MessageAndOffset V1.
For producer, there will be no impact.
In phase 1, there will be no impact for consumers.
In phase 2, there will be some performance penalty for consumers that only supports MessageAndOffset V0, because there is no zero-copy transfer.
At the beginning of phase 2, there will be some time the log segment contains both MessageAndOffset V0 and V1. The broker will always do down conversion for FetchRequest V1 and zero-copy transfer for FetchRequest V2.
** We introduce internal ApiVersion here to help the user who are running on trunk to upgrade in the future. Otherwise the interim ApiVersion between two official releases will require users to downgrade ApiVersion then upgrade.
To canary a broker
After phase 1, it is possible for user to canary a broker in phase 2 and roll back if something goes wrong. The procedure is:
- Set message.format.version=1 on one of the brokers (broker B).
- Broker B will start to act like what described in phase 2.
- It will sends FetchRequest V2 to other brokers for replication.
- It will only see ProduceRequest/FetchRequest V1 from other brokers and clietns.
- If something goes wrong, we can do the following to rollback:
- shutdown broker B
- nuke the data of the topics it was serving as leader before shutdown
- set message.format.version=0
- restart the broker to let the broker replicate from leaders. At this point the data on disk will be in MessageAndOffset V0.
- shutdown broker B
In step 2, it is recommended to put only small amount of leaders on the broker, because at that point the broker needs to do down conversion for all the fetch requests.
Option discussions with use cases
For documentation purpose, here are some discussions we had on this KIP.
This section discusses how the three options work with a few use cases. Option 1 and Option 2 are in the rejected option section.
Options comparison
| Use cases | Option 1 (Message contains CreateTime + LogAppendTime) | option 2 (Message contains LogAppendTime only) | option 3 (message contains CreateTime only, brokers keep LogAppendTime in log index) | option 4 (Message contains a timestamp that could be overridden by broker depending on configured time difference threshold) | Comparison |
|---|---|---|---|---|---|
| Mirror Maker | Broker overrides the LAT and keep the CT as is. | Broker overrides the LAT | Broker keep the CT as. And add index entry with LAT to the log index file. | Mirror maker will keep the consumed messages' timestamp. Those timestamp may or may not be overwritten by target cluster depending on the configuration. | Option 1 provides the most information to user. The only concern is whether we should expose LAT to user. Option 2 loses the CreateTime information. Option 3 have same amount information as option 1 from broker point of view. From user point of view, it does not expose the LAT. Option 4 could be equivalent to either having CreateTime only or having LogAppendTime only depending on configuration. |
| Log Retention | Broker will use the LAT of the last message in a segment to enforce the policy. | Same as option 1. | Broker will use the LAT of the last entry in the log index file to enforce the retention policy. Because the leader is the source of truth for LAT, followers need to get the LAT from leader when they replicate the messages. That means we need to introduce a new wire protocol to fetch the time based log index file as well. When log recovery happens, the rebuilt time index would have different LAT from the actual arrival time of the messages in the log. And the LAT in the index file will be very close, or even the same. | The broker will take a look at the last Time Index entry of the segment to decide whether to delete the log segment or not. | Option 1 and option 2 can work with existing replication design and solve the log retention issue we have now. Option 3 solves the issue but is a bit involved because it needs to replicate the time index entry as well. Option 4 could be equivalent to either having CreateTime only or having LogAppendTime only depending on configuration. |
| Log rolling | Broker will use the LAT of the first message in a log segment to enforce the policy. | Same as option 1. | Broker will use the LAT of the first entry in the log index file to enforce the retention policy. Similar to the log retention case, the followers needs to replicate the time index as well. The log recovery happens, the log rolling might not be honored either. | The broker will keep an in memory earliest message timestamp in the active log segment. On broker startup, the broker will need to scan the messages in the active segment to find the earliest timestamp. | Option 1 and option 2 solves the log rolling issue. Option 3 solves the issue but is a bit involved because it needs to replicate the time index entry as well. On broker start up, Option 4 needs to scan the active log segment to find the earliest timestamp in the log segment. |
| Stream processing | Applications don't need to include the CreateTime in the payload but simply use the CreateTime field. | Applications have to put CreateTime into the payload. | Applications don't need to include the CreateTime in the payload but simply use the CreateTime field. | Option 4 could be equivalent to either having CreateTime only or having LogAppendTime only depending on configuration. | The benefit of having a CreateTime with each message rather than put it into payload is that application protocol can be simplified. It is convenient for the infrastructure to provide the timestamp so there is no need for each application to worry about the timestamp. |
| Latency measurement | User can get End2End latency and lag in time. | User can get the lag in time. | User can get End2End latency. | Depending on the max.message.time.difference.ms configuration. User may or may not be able to find out the latency. | Option 1 has most information for user. |
| Search message by timestamp. | Detail discussion in KIP-33 | Detail discussion in KIP-33 | Detail discussion in KIP-33 | Detail discussion in KIP-33 | Detail discussion in KIP-33 |
Mirror maker case in detail
The behavior of broker for all the options are the same: The broker will always override the LogAppendTime(if exists) when message arrives the broker and keep the CreateTime(if exists) untouched.
The broker does not distinguish mirror maker from other producers. The following example explains what will the timestamp look like when there is mirror maker in the picture.(CT - CreateTime, LAT - LogAppendTime)
- Application producer produces message at T0. ( [CT = T0, LAT = -1] )
- The broker in cluster1 received message at T1 = T0 + latency1 and appended the message to the log. Latency1 includes linger.ms and some other latency. ( [CT = T0, LAT = T1] )
- Mirror maker copies the message to broker in cluster2. ( [CT = T0, LAT = T1] )
- The broker in cluster2 receives message at T2 = T1 + latency2 and appended the message to the log. ( [CT = T0, LAT = T2] )
The CreateTime of a message in source cluster and target cluster will be same. i.e. the timestamp is passed across clusters.
The LogAppendTime of a message in source cluster and target cluster will be different.
Discussion: should we use CreateTime OR LogAppendTime for log retention and time based log rolling?
To discuss the usage of CreateTime and LogAppendTime, it is useful to summarize the latency pattern of the messages flowing through the pipeline. The latency can be summarized to the following pattern:
- the messages flow through the pipeline with small latency.
- the messages flow through the pipeline with similar large latency.
- the messages flow through the pipeline with large latency difference.
Also it would be useful to think about the impact of a completely wrong timestamp set by client. i.e. the robustness of the system.
Log Retention
There are both pros and cons for log retention to be based on CreateTime or LogAppendTime.
- Pattern 1:
Because the latency is small, so CreateTime and LogAppendTime will be close and their won't be much difference. - pattern 2:
If the log retention is based on the message creation time, it will not be affected by the latency in the data pipeline because the send time will not change.
If the log retention is based on the LogAppendTime, it will be affected by the latency in the pipeline. Because of the latency difference, some data can be deleted on one cluster, but not on another cluster in the pipeline. - Pattern 3:
When the messages with significantly different timestamp goes into a cluster at around same time, the retention policy is hard to follow if we use CreateTime. For example, imagine two mirror makers copy data from two source clusters to the same target cluster. If MirrorMaker1 is copying Messages with CreateTime around 1:00 PM today, and MirrorMaker2 is copying messages with CreateTime around 1:00 PM yesterday. Those messages can go to the same log segment in the target cluster. It will be difficult for broker to apply retention policy to the log segment. The broker needs to maintain the knowledge about the latest CreateTime of all messages in a log segment and persist the information somewhere. - Robustness:
If there is a message with CreateTime set to the future, the log might be kept for very long. Broker needs to sanity check the timestamp when receive the message. It could by tricky to determine which timestamp is not valid
Comparison:
| pattern 1 | pattern 2 | pattern 3 | Robustness | |
|---|---|---|---|---|
| Preference | CT or LAT | CT | LAT | LAT |
In reality, we usually don't see all the pipeline has same large latency, so it looks LogAppendTime is preferable than CreateTime for log retention.
Time based log rolling
The main purpose of time based log rolling is to avoid the situation where a low volume topic always has only one segment which is also the active segment. From its nature, server side log rolling makes more sense.
- Pattern 1:
Because the latency is small, so CreateTime and LogAppendTime will be close and their won't be much difference. - Pattern 2:
When the latency is large, it is possible that when a new message is produced to a broker, the CreateTime has already reached the rolling criteria. This might cause a segment with only one message. - Pattern 3:
Similar to pattern 2, a lagged message might result in a single message log segment. - Robustness:
Similar to as pattern 2 and pattern 3. Also a CreateTime in the future might break the log rolling as well.
| pattern 1 | pattern 2 | pattern 3 | Robustness | |
|---|---|---|---|---|
| Preference | CT or LAT | LAT | LAT | LAT |
Rejected Alternatives
Option 1 - Add both CreateTime and LogAppendTime to message format
...