You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/07/16 17:37:51 UTC
[GitHub] [kafka] hachikuji commented on a change in pull request #11036: KAFKA-12944: Assume message format version is 3.0 when inter-broker protocol is 3.0 or higher (KIP-724)
hachikuji commented on a change in pull request #11036:
URL: https://github.com/apache/kafka/pull/11036#discussion_r671401329
##########
File path: clients/src/main/java/org/apache/kafka/common/config/TopicConfig.java
##########
@@ -171,12 +171,24 @@
public static final String PREALLOCATE_DOC = "True if we should preallocate the file on disk when " +
"creating a new log segment.";
+ /**
+ * @deprecated since 3.0, removal planned in 4.0. The default value for this config is appropriate
+ * for most situations.
+ */
+ @Deprecated
public static final String MESSAGE_FORMAT_VERSION_CONFIG = "message.format.version";
- public static final String MESSAGE_FORMAT_VERSION_DOC = "Specify the message format version the broker " +
- "will use to append messages to the logs. The value should be a valid ApiVersion. Some examples are: " +
- "0.8.2, 0.9.0.0, 0.10.0, check ApiVersion for more details. By setting a particular message format " +
- "version, the user is certifying that all the existing messages on disk are smaller or equal than the " +
- "specified version. Setting this value incorrectly will cause consumers with older versions to break as " +
+
+ /**
+ * @deprecated since 3.0, removal planned in 4.0. The default value for this config is appropriate
+ * for most situations.
+ */
+ @Deprecated
+ public static final String MESSAGE_FORMAT_VERSION_DOC = "[DEPRECATED] Specify the message format version the broker " +
+ "will use to append messages to the logs if inter-broker. The value of this config is always assumed to be `3.0` if " +
Review comment:
Remove "if inter-broker" in the first sentence?
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -774,10 +774,10 @@ class KafkaApis(val requestChannel: RequestChannel,
// which were written in the new format prior to the version downgrade.
val unconvertedRecords = FetchResponse.recordsOrFail(partitionData)
val downConvertMagic =
- logConfig.map(_.messageFormatVersion.recordVersion.value).flatMap { magic =>
- if (magic > RecordBatch.MAGIC_VALUE_V0 && versionId <= 1 && !unconvertedRecords.hasCompatibleMagic(RecordBatch.MAGIC_VALUE_V0))
+ logConfig.map(_.recordVersion.value).flatMap { magic =>
Review comment:
Hmm, I thought we were going to simplify this. Basically ignore the log config and just rely on the fetch version. The intuition is that _all_ new data will need to be down-converted anyway. Perhaps we want to wait until the IBP is at 3.0 for this logic to kick in?
##########
File path: clients/src/main/java/org/apache/kafka/common/config/TopicConfig.java
##########
@@ -171,12 +171,24 @@
public static final String PREALLOCATE_DOC = "True if we should preallocate the file on disk when " +
"creating a new log segment.";
+ /**
+ * @deprecated since 3.0, removal planned in 4.0. The default value for this config is appropriate
+ * for most situations.
+ */
+ @Deprecated
public static final String MESSAGE_FORMAT_VERSION_CONFIG = "message.format.version";
- public static final String MESSAGE_FORMAT_VERSION_DOC = "Specify the message format version the broker " +
- "will use to append messages to the logs. The value should be a valid ApiVersion. Some examples are: " +
- "0.8.2, 0.9.0.0, 0.10.0, check ApiVersion for more details. By setting a particular message format " +
- "version, the user is certifying that all the existing messages on disk are smaller or equal than the " +
- "specified version. Setting this value incorrectly will cause consumers with older versions to break as " +
+
+ /**
+ * @deprecated since 3.0, removal planned in 4.0. The default value for this config is appropriate
+ * for most situations.
+ */
+ @Deprecated
+ public static final String MESSAGE_FORMAT_VERSION_DOC = "[DEPRECATED] Specify the message format version the broker " +
+ "will use to append messages to the logs if inter-broker. The value of this config is always assumed to be `3.0` if " +
+ "`inter.broker.protocol.version` is 3.0 or higher (the actual config value is ignored). Otherwise, the value should" +
+ "be a valid ApiVersion. Some examples are: 0.8.2, 0.9.0.0, 0.10.0, check ApiVersion for more details. By setting a " +
Review comment:
No need to fix it here since this has been here forever, but it's a little weird for the doc to mention a classname. Will users have any idea what that means?
##########
File path: clients/src/main/java/org/apache/kafka/common/config/TopicConfig.java
##########
@@ -171,12 +171,24 @@
public static final String PREALLOCATE_DOC = "True if we should preallocate the file on disk when " +
"creating a new log segment.";
+ /**
+ * @deprecated since 3.0, removal planned in 4.0. The default value for this config is appropriate
+ * for most situations.
+ */
+ @Deprecated
public static final String MESSAGE_FORMAT_VERSION_CONFIG = "message.format.version";
- public static final String MESSAGE_FORMAT_VERSION_DOC = "Specify the message format version the broker " +
- "will use to append messages to the logs. The value should be a valid ApiVersion. Some examples are: " +
- "0.8.2, 0.9.0.0, 0.10.0, check ApiVersion for more details. By setting a particular message format " +
- "version, the user is certifying that all the existing messages on disk are smaller or equal than the " +
- "specified version. Setting this value incorrectly will cause consumers with older versions to break as " +
+
+ /**
+ * @deprecated since 3.0, removal planned in 4.0. The default value for this config is appropriate
+ * for most situations.
+ */
+ @Deprecated
+ public static final String MESSAGE_FORMAT_VERSION_DOC = "[DEPRECATED] Specify the message format version the broker " +
+ "will use to append messages to the logs if inter-broker. The value of this config is always assumed to be `3.0` if " +
+ "`inter.broker.protocol.version` is 3.0 or higher (the actual config value is ignored). Otherwise, the value should" +
Review comment:
nit: missing space at the end
##########
File path: core/src/main/scala/kafka/admin/ConfigCommand.scala
##########
@@ -270,7 +271,8 @@ object ConfigCommand extends Config {
}
if (props.containsKey(LogConfig.MessageFormatVersionProp)) {
println(s"WARNING: The configuration ${LogConfig.MessageFormatVersionProp}=${props.getProperty(LogConfig.MessageFormatVersionProp)} is specified. " +
- s"This configuration will be ignored if the version is newer than the inter.broker.protocol.version specified in the broker.")
+ "This configuration will be ignored if the version is newer than the inter.broker.protocol.version specified in the broker or " +
Review comment:
Should we also mention that the configuration is deprecated? Same question `TopicCommand` and the warning message in `LogConfig`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org