MINOR: Ensure timestamp type is provided when up-converting messages #2483
Conversation
|
cc @ijuma I ran into a couple issues with up-conversion when working with EoS message format changes. Seems we should be a little safer with the handling of NO_TIMESTAMP_TYPE. |
hachikuji
changed the title
|
Refer to this link for build results (access rights to CI server needed): |
|
Refer to this link for build results (access rights to CI server needed): |
|
Refer to this link for build results (access rights to CI server needed): |
| @@ -316,7 +316,7 @@ public void testConvertNonCompressedToMagic1() throws IOException { | |||
| try (FileRecords fileRecords = FileRecords.open(tempFile())) { | |||
| fileRecords.append(records); | |||
| fileRecords.flush(); | |||
| Records convertedRecords = fileRecords.toMessageFormat(Record.MAGIC_VALUE_V1); | |||
| Records convertedRecords = fileRecords.toMessageFormat(Record.MAGIC_VALUE_V1, TimestampType.CREATE_TIME); | |||
There was a problem hiding this comment.
Is there a reason for picking CREATE_TIME instead of LOG_APPEND_TIME? I guess either is wrong since the timestamp field will be -1 in this case, right?
There was a problem hiding this comment.
No particular reason. We have to choose one of them when up-converting. This just verifies that it uses what we set.
| def testCreateTimeUpConversion() { | ||
| val records = createRecords(magicValue = Record.MAGIC_VALUE_V0, codec = CompressionType.GZIP) | ||
| val validatedResults = | ||
| LogValidator.validateMessagesAndAssignOffsets(records, |
There was a problem hiding this comment.
Nit: shouldn't this be in the previous line?
| @@ -186,7 +186,7 @@ private[kafka] object LogValidator { | |||
| if (record.magic != messageFormatVersion) | |||
| inPlaceAssignment = false | |||
|
|
|||
| validatedRecords += record.convert(messageFormatVersion) | |||
| validatedRecords += record.convert(messageFormatVersion, messageTimestampType) | |||
There was a problem hiding this comment.
Is messageTimestampType never NO_TIMESTAMP if messageFormatVersion > 0?
There was a problem hiding this comment.
Good question. It seems it may actually be possible for the user to use "NO_TIMESTAMP_TYPE" for "message.timestamp.type." If so, it's probably worth checking for that and raising an exception.
There was a problem hiding this comment.
Looked into this. Our config definition constrains message.timestamp.type to be one of "CreateTime" or "LogAppendTime," so I think we are safe.
| throw new IllegalArgumentException("Cannot up-convert using timestamp type " + upconvertTimestampType); | ||
| timestampType = upconvertTimestampType; | ||
| } else { | ||
| timestampType = TimestampType.forAttributes(attributes()); |
There was a problem hiding this comment.
How come we don't check wrapperRecordTimestampType any more (which could have been done by simply calling timestampType() btw)?
Also, can you search for timstamp in this file and fix an existing typo?
There was a problem hiding this comment.
Ah, you are right. I'll fix. Hmm... wonder why no tests failed.
There was a problem hiding this comment.
Oh, I guess it's because we would never have a wrapper timestamp type if we're converting from magic 0 to magic 1, and of course it wouldn't matter when converting in the other direction. But we will need to consider it when up-converting to magic 2.
hachikuji
force-pushed
the
minor-upconvert-timestamp-safety
branch
from
1e0d210
to
b42804b
Compare
|
Refer to this link for build results (access rights to CI server needed): |
|
Refer to this link for build results (access rights to CI server needed): |
|
Refer to this link for build results (access rights to CI server needed): |
Author: Jason Gustafson <jason@confluent.io> Reviewers: Ismael Juma <ismael@juma.me.uk> Closes apache#2483 from hachikuji/minor-upconvert-timestamp-safety
No description provided.