KAFKA-3047: Explicit offset assignment in Log.append can corrupt the log

This fix was suggested by Maciek Makowski, who also reported the problem.

Author: Ismael Juma <ismael@juma.me.uk>

Reviewers: Guozhang Wang <wangguoz@gmail.com>

Closes #1029 from ijuma/KAFKA-3047-log-append-can-corrupt-the-log

core/src/main/scala/kafka/log/Log.scala

@@ -328,11 +328,11 @@ class Log(val dir: File,
     try {
       // they are valid, insert them in the log
       lock synchronized {
-        appendInfo.firstOffset = nextOffsetMetadata.messageOffset
 
         if (assignOffsets) {
           // assign offsets to the message set
           val offset = new LongRef(nextOffsetMetadata.messageOffset)
+          appendInfo.firstOffset = offset.value
           val now = time.milliseconds
           val (validatedMessages, messageSizesMaybeChanged) = try {
             validMessages.validateMessagesAndAssignOffsets(offset,