Repository: kafka Updated Branches: refs/heads/0.10.0 625e94001 -> 6500b53c7
KAFKA-3787; Preserve the message timestamp in mirror maker The timestamp of messages consumed by mirror maker is not preserved after sending to target cluster. The correct behavior is to keep create timestamp the same in both source and target clusters. Author: Tao Xiao <[email protected]> Reviewers: Jun Rao <[email protected]>, Ismael Juma <[email protected]> Closes #1466 from xiaotao183/KAFKA-3787 (cherry picked from commit f4a263b5a89e028ad3b658dca2259b6da0197214) Signed-off-by: Ismael Juma <[email protected]> Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/6500b53c Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/6500b53c Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/6500b53c Branch: refs/heads/0.10.0 Commit: 6500b53c75495296275e83e4e4beb224d7ac0c4f Parents: 625e940 Author: Tao Xiao <[email protected]> Authored: Sun Jun 5 08:42:42 2016 +0100 Committer: Ismael Juma <[email protected]> Committed: Sun Jun 5 08:58:06 2016 +0100 ---------------------------------------------------------------------- .../main/scala/kafka/tools/MirrorMaker.scala | 4 +- .../unit/kafka/tools/MirrorMakerTest.scala | 42 ++++++++++++++++++++ 2 files changed, 44 insertions(+), 2 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kafka/blob/6500b53c/core/src/main/scala/kafka/tools/MirrorMaker.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/tools/MirrorMaker.scala b/core/src/main/scala/kafka/tools/MirrorMaker.scala index 87f3cc5..7d6b5fb 100755 --- a/core/src/main/scala/kafka/tools/MirrorMaker.scala +++ b/core/src/main/scala/kafka/tools/MirrorMaker.scala @@ -673,9 +673,9 @@ object MirrorMaker extends Logging with KafkaMetricsGroup { def handle(record: BaseConsumerRecord): util.List[ProducerRecord[Array[Byte], Array[Byte]]] } - private object defaultMirrorMakerMessageHandler extends MirrorMakerMessageHandler { + private[tools] object defaultMirrorMakerMessageHandler extends MirrorMakerMessageHandler { override def handle(record: BaseConsumerRecord): util.List[ProducerRecord[Array[Byte], Array[Byte]]] = { - Collections.singletonList(new ProducerRecord[Array[Byte], Array[Byte]](record.topic, record.key, record.value)) + Collections.singletonList(new ProducerRecord[Array[Byte], Array[Byte]](record.topic, null, record.timestamp, record.key, record.value)) } } http://git-wip-us.apache.org/repos/asf/kafka/blob/6500b53c/core/src/test/scala/unit/kafka/tools/MirrorMakerTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/unit/kafka/tools/MirrorMakerTest.scala b/core/src/test/scala/unit/kafka/tools/MirrorMakerTest.scala new file mode 100644 index 0000000..39a0ac9 --- /dev/null +++ b/core/src/test/scala/unit/kafka/tools/MirrorMakerTest.scala @@ -0,0 +1,42 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package kafka.tools + +import kafka.consumer.BaseConsumerRecord +import org.apache.kafka.common.record.TimestampType +import org.junit.Assert._ +import org.junit.Test + +class MirrorMakerTest { + + @Test + def testDefaultMirrorMakerMessageHandler() { + val now = 12345L + val consumerRecord = BaseConsumerRecord("topic", 0, 1L, now, TimestampType.CREATE_TIME, "key".getBytes, "value".getBytes) + + val result = MirrorMaker.defaultMirrorMakerMessageHandler.handle(consumerRecord) + assertEquals(1, result.size) + + val producerRecord = result.get(0) + assertEquals(now, producerRecord.timestamp) + assertEquals("topic", producerRecord.topic) + assertNull(producerRecord.partition) + assertEquals("key", new String(producerRecord.key)) + assertEquals("value", new String(producerRecord.value)) + } +}
