Repository: kafka Updated Branches: refs/heads/trunk 23f695196 -> 8b3c6c0a1
KAFKA-4597; RecordMetadata should return log append time when appropriate There is a slight change of behaviour: we now complete the `Future` returned from `send` before the callbacks are invoked. This seems OK and perhaps a little better as the `Future` can make progress sooner (as it would typically be blocked on a different thread than the I/O thread that invokes the callbacks). Author: Ismael Juma <[email protected]> Reviewers: Ewen Cheslack-Postava <[email protected]>, Jason Gustafson <[email protected]> Closes #2318 from ijuma/kafka-4597-record-metadata-log-append-time Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/8b3c6c0a Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/8b3c6c0a Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/8b3c6c0a Branch: refs/heads/trunk Commit: 8b3c6c0a1c1bb8f098a7013dbf9c2cb97720e106 Parents: 23f6951 Author: Ismael Juma <[email protected]> Authored: Wed Jan 25 17:25:10 2017 -0800 Committer: Jason Gustafson <[email protected]> Committed: Wed Jan 25 17:25:10 2017 -0800 ---------------------------------------------------------------------- .../kafka/clients/producer/MockProducer.java | 13 ++--- .../internals/FutureRecordMetadata.java | 38 ++++---------- .../internals/ProduceRequestResult.java | 37 ++++++++++--- .../clients/producer/internals/RecordBatch.java | 55 +++++++++----------- .../clients/producer/internals/Sender.java | 25 +++++---- .../apache/kafka/common/protocol/Protocol.java | 2 +- .../kafka/common/requests/ProduceRequest.java | 7 ++- .../kafka/common/requests/ProduceResponse.java | 36 +++++++------ .../clients/producer/KafkaProducerTest.java | 15 ++++++ .../kafka/clients/producer/RecordSendTest.java | 18 +++---- .../clients/producer/internals/SenderTest.java | 3 +- .../common/requests/RequestResponseTest.java | 6 ++- .../coordinator/GroupMetadataManager.scala | 18 +++---- .../main/scala/kafka/server/DelayedFetch.scala | 2 +- .../scala/kafka/server/DelayedProduce.scala | 8 +-- .../src/main/scala/kafka/server/KafkaApis.scala | 14 +++-- .../scala/kafka/server/ReplicaManager.scala | 41 +++++++-------- .../kafka/api/AuthorizerIntegrationTest.scala | 2 +- .../kafka/api/BaseProducerSendTest.scala | 23 +++++--- .../kafka/api/PlaintextProducerSendTest.scala | 54 +++---------------- .../GroupCoordinatorResponseTest.scala | 4 +- .../coordinator/GroupMetadataManagerTest.scala | 2 +- .../unit/kafka/server/EdgeCaseRequestTest.scala | 4 +- .../unit/kafka/server/ProduceRequestTest.scala | 8 +-- .../unit/kafka/server/ReplicaManagerTest.scala | 10 ++-- 25 files changed, 217 insertions(+), 228 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java index 109b0ca..008ca70 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java @@ -116,11 +116,11 @@ public class MockProducer<K, V> implements Producer<K, V> { int partition = 0; if (this.cluster.partitionsForTopic(record.topic()) != null) partition = partition(record, this.cluster); - ProduceRequestResult result = new ProduceRequestResult(); - FutureRecordMetadata future = new FutureRecordMetadata(result, 0, Record.NO_TIMESTAMP, 0, 0, 0); TopicPartition topicPartition = new TopicPartition(record.topic(), partition); + ProduceRequestResult result = new ProduceRequestResult(topicPartition); + FutureRecordMetadata future = new FutureRecordMetadata(result, 0, Record.NO_TIMESTAMP, 0, 0, 0); long offset = nextOffset(topicPartition); - Completion completion = new Completion(topicPartition, offset, + Completion completion = new Completion(offset, new RecordMetadata(topicPartition, 0, offset, Record.NO_TIMESTAMP, 0, 0, 0), result, callback); this.sent.add(record); @@ -233,10 +233,8 @@ public class MockProducer<K, V> implements Producer<K, V> { private final RecordMetadata metadata; private final ProduceRequestResult result; private final Callback callback; - private final TopicPartition topicPartition; - public Completion(TopicPartition topicPartition, - long offset, + public Completion(long offset, RecordMetadata metadata, ProduceRequestResult result, Callback callback) { @@ -244,11 +242,10 @@ public class MockProducer<K, V> implements Producer<K, V> { this.offset = offset; this.result = result; this.callback = callback; - this.topicPartition = topicPartition; } public void complete(RuntimeException e) { - result.done(topicPartition, e == null ? offset : -1L, e); + result.done(e == null ? offset : -1L, Record.NO_TIMESTAMP, e); if (callback != null) { if (e == null) callback.onCompletion(metadata, null); http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/clients/src/main/java/org/apache/kafka/clients/producer/internals/FutureRecordMetadata.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/FutureRecordMetadata.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/FutureRecordMetadata.java index d5995a3..1cde13d 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/FutureRecordMetadata.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/FutureRecordMetadata.java @@ -26,16 +26,16 @@ public final class FutureRecordMetadata implements Future<RecordMetadata> { private final ProduceRequestResult result; private final long relativeOffset; - private final long timestamp; + private final long createTimestamp; private final long checksum; private final int serializedKeySize; private final int serializedValueSize; - public FutureRecordMetadata(ProduceRequestResult result, long relativeOffset, long timestamp, + public FutureRecordMetadata(ProduceRequestResult result, long relativeOffset, long createTimestamp, long checksum, int serializedKeySize, int serializedValueSize) { this.result = result; this.relativeOffset = relativeOffset; - this.timestamp = timestamp; + this.createTimestamp = createTimestamp; this.checksum = checksum; this.serializedKeySize = serializedKeySize; this.serializedValueSize = serializedValueSize; @@ -47,6 +47,11 @@ public final class FutureRecordMetadata implements Future<RecordMetadata> { } @Override + public boolean isCancelled() { + return false; + } + + @Override public RecordMetadata get() throws InterruptedException, ExecutionException { this.result.await(); return valueOrError(); @@ -69,32 +74,11 @@ public final class FutureRecordMetadata implements Future<RecordMetadata> { RecordMetadata value() { return new RecordMetadata(result.topicPartition(), this.result.baseOffset(), this.relativeOffset, - this.timestamp, this.checksum, this.serializedKeySize, this.serializedValueSize); - } - - public long relativeOffset() { - return this.relativeOffset; - } - - public long timestamp() { - return this.timestamp; + timestamp(), this.checksum, this.serializedKeySize, this.serializedValueSize); } - public long checksum() { - return this.checksum; - } - - public int serializedKeySize() { - return this.serializedKeySize; - } - - public int serializedValueSize() { - return this.serializedValueSize; - } - - @Override - public boolean isCancelled() { - return false; + private long timestamp() { + return result.hasLogAppendTime() ? result.logAppendTime() : createTimestamp; } @Override http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProduceRequestResult.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProduceRequestResult.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProduceRequestResult.java index 8e5855d..41c204d 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProduceRequestResult.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProduceRequestResult.java @@ -19,7 +19,9 @@ package org.apache.kafka.clients.producer.internals; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.record.Record; /** @@ -30,22 +32,31 @@ import org.apache.kafka.common.TopicPartition; public final class ProduceRequestResult { private final CountDownLatch latch = new CountDownLatch(1); - private volatile TopicPartition topicPartition; + private final TopicPartition topicPartition; + private volatile long baseOffset = -1L; + private volatile long logAppendTime = Record.NO_TIMESTAMP; private volatile RuntimeException error; - public ProduceRequestResult() { + /** + * Create an instance of this class. + * + * @param topicPartition The topic and partition to which this record set was sent was sent + */ + public ProduceRequestResult(TopicPartition topicPartition) { + this.topicPartition = topicPartition; } /** * Mark this request as complete and unblock any threads waiting on its completion. - * @param topicPartition The topic and partition to which this record set was sent was sent + * * @param baseOffset The base offset assigned to the record - * @param error The error that occurred if there was one, or null. + * @param logAppendTime The log append time or -1 if CreateTime is being used + * @param error The error that occurred if there was one, or null */ - public void done(TopicPartition topicPartition, long baseOffset, RuntimeException error) { - this.topicPartition = topicPartition; + public void done(long baseOffset, long logAppendTime, RuntimeException error) { this.baseOffset = baseOffset; + this.logAppendTime = logAppendTime; this.error = error; this.latch.countDown(); } @@ -75,6 +86,20 @@ public final class ProduceRequestResult { } /** + * Return true if log append time is being used for this topic + */ + public boolean hasLogAppendTime() { + return logAppendTime != Record.NO_TIMESTAMP; + } + + /** + * The log append time or -1 if CreateTime is being used + */ + public long logAppendTime() { + return logAppendTime; + } + + /** * The error thrown (generally on the server) while processing this request */ public RuntimeException error() { http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordBatch.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordBatch.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordBatch.java index 68b27d3..1b751bf 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordBatch.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordBatch.java @@ -35,28 +35,28 @@ public final class RecordBatch { private static final Logger log = LoggerFactory.getLogger(RecordBatch.class); - public int recordCount = 0; - public int maxRecordSize = 0; - public volatile int attempts = 0; - public final long createdMs; - public long drainedMs; - public long lastAttemptMs; - public final TopicPartition topicPartition; - public final ProduceRequestResult produceFuture; - public long lastAppendTime; - private final List<Thunk> thunks; - private boolean retry; + final long createdMs; + final TopicPartition topicPartition; + final ProduceRequestResult produceFuture; + + private final List<Thunk> thunks = new ArrayList<>(); private final MemoryRecordsBuilder recordsBuilder; + volatile int attempts; + int recordCount; + int maxRecordSize; + long drainedMs; + long lastAttemptMs; + long lastAppendTime; + private boolean retry; + public RecordBatch(TopicPartition tp, MemoryRecordsBuilder recordsBuilder, long now) { this.createdMs = now; this.lastAttemptMs = now; this.recordsBuilder = recordsBuilder; this.topicPartition = tp; - this.produceFuture = new ProduceRequestResult(); - this.thunks = new ArrayList<>(); this.lastAppendTime = createdMs; - this.retry = false; + this.produceFuture = new ProduceRequestResult(topicPartition); } /** @@ -83,36 +83,33 @@ public final class RecordBatch { } /** - * Complete the request + * Complete the request. * * @param baseOffset The base offset of the messages assigned by the server - * @param timestamp The timestamp returned by the broker. + * @param logAppendTime The log append time or -1 if CreateTime is being used * @param exception The exception that occurred (or null if the request was successful) */ - public void done(long baseOffset, long timestamp, RuntimeException exception) { + public void done(long baseOffset, long logAppendTime, RuntimeException exception) { log.trace("Produced messages to topic-partition {} with base offset offset {} and error: {}.", - topicPartition, - baseOffset, - exception); + topicPartition, baseOffset, exception); + + // Complete the future before invoking the callbacks as we rely on its state for the `onCompletion` call + produceFuture.done(baseOffset, logAppendTime, exception); + // execute callbacks for (Thunk thunk : thunks) { try { if (exception == null) { - // If the timestamp returned by server is NoTimestamp, that means CreateTime is used. Otherwise LogAppendTime is used. - RecordMetadata metadata = new RecordMetadata(this.topicPartition, baseOffset, thunk.future.relativeOffset(), - timestamp == Record.NO_TIMESTAMP ? thunk.future.timestamp() : timestamp, - thunk.future.checksum(), - thunk.future.serializedKeySize(), - thunk.future.serializedValueSize()); + RecordMetadata metadata = thunk.future.value(); thunk.callback.onCompletion(metadata, null); } else { thunk.callback.onCompletion(null, exception); } } catch (Exception e) { - log.error("Error executing user-provided callback on message for topic-partition {}:", topicPartition, e); + log.error("Error executing user-provided callback on message for topic-partition '{}'", topicPartition, e); } } - this.produceFuture.done(topicPartition, baseOffset, exception); + } /** @@ -167,7 +164,7 @@ public final class RecordBatch { /** * Returns if the batch is been retried for sending to kafka */ - public boolean inRetry() { + private boolean inRetry() { return this.retry; } http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java index dc29936..5483c4a 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java @@ -34,7 +34,6 @@ import org.apache.kafka.common.metrics.stats.Max; import org.apache.kafka.common.metrics.stats.Rate; import org.apache.kafka.common.protocol.Errors; import org.apache.kafka.common.record.MemoryRecords; -import org.apache.kafka.common.record.Record; import org.apache.kafka.common.requests.ProduceRequest; import org.apache.kafka.common.requests.ProduceResponse; import org.apache.kafka.common.utils.Time; @@ -253,12 +252,12 @@ public class Sender implements Runnable { if (response.wasDisconnected()) { log.trace("Cancelled request {} due to node {} being disconnected", response, response.destination()); for (RecordBatch batch : batches.values()) - completeBatch(batch, Errors.NETWORK_EXCEPTION, -1L, Record.NO_TIMESTAMP, correlationId, now); + completeBatch(batch, new ProduceResponse.PartitionResponse(Errors.NETWORK_EXCEPTION), correlationId, now); } else if (response.versionMismatch() != null) { log.warn("Cancelled request {} due to a version mismatch with node {}", response, response.destination(), response.versionMismatch()); for (RecordBatch batch : batches.values()) - completeBatch(batch, Errors.INVALID_REQUEST, -1L, Record.NO_TIMESTAMP, correlationId, now); + completeBatch(batch, new ProduceResponse.PartitionResponse(Errors.INVALID_REQUEST), correlationId, now); } else { log.trace("Received produce response from node {} with correlation id {}", response.destination(), correlationId); // if we have a response, parse it @@ -267,16 +266,16 @@ public class Sender implements Runnable { for (Map.Entry<TopicPartition, ProduceResponse.PartitionResponse> entry : produceResponse.responses().entrySet()) { TopicPartition tp = entry.getKey(); ProduceResponse.PartitionResponse partResp = entry.getValue(); - Errors error = Errors.forCode(partResp.errorCode); RecordBatch batch = batches.get(tp); - completeBatch(batch, error, partResp.baseOffset, partResp.timestamp, correlationId, now); + completeBatch(batch, partResp, correlationId, now); } this.sensors.recordLatency(response.destination(), response.requestLatencyMs()); this.sensors.recordThrottleTime(produceResponse.getThrottleTime()); } else { // this is the acks = 0 case, just complete all requests - for (RecordBatch batch : batches.values()) - completeBatch(batch, Errors.NONE, -1L, Record.NO_TIMESTAMP, correlationId, now); + for (RecordBatch batch : batches.values()) { + completeBatch(batch, new ProduceResponse.PartitionResponse(Errors.NONE), correlationId, now); + } } } } @@ -285,13 +284,13 @@ public class Sender implements Runnable { * Complete or retry the given batch of records. * * @param batch The record batch - * @param error The error (or null if none) - * @param baseOffset The base offset assigned to the records if successful - * @param timestamp The timestamp returned by the broker for this batch + * @param response The produce response * @param correlationId The correlation id for the request - * @param now The current POSIX time stamp in milliseconds + * @param now The current POSIX timestamp in milliseconds */ - private void completeBatch(RecordBatch batch, Errors error, long baseOffset, long timestamp, long correlationId, long now) { + private void completeBatch(RecordBatch batch, ProduceResponse.PartitionResponse response, long correlationId, + long now) { + Errors error = response.error; if (error != Errors.NONE && canRetry(batch, error)) { // retry log.warn("Got error produce response with correlation id {} on topic-partition {}, retrying ({} attempts left). Error: {}", @@ -308,7 +307,7 @@ public class Sender implements Runnable { else exception = error.exception(); // tell the user the result of their request - batch.done(baseOffset, timestamp, exception); + batch.done(response.baseOffset, response.logAppendTime, exception); this.accumulator.deallocate(batch); if (error != Errors.NONE) this.sensors.recordErrors(batch.topicPartition.topic(), batch.recordCount); http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java b/clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java index e76fc04..b04587f 100644 --- a/clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java +++ b/clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java @@ -193,7 +193,7 @@ public class Protocol { INT16), new Field("base_offset", INT64), - new Field("timestamp", + new Field("log_append_time", INT64, "The timestamp returned by broker after appending the messages. " + "If CreateTime is used for the topic, the timestamp will be -1. " + http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java index c05643d..e4a2aae 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java @@ -20,7 +20,6 @@ import org.apache.kafka.common.protocol.Errors; import org.apache.kafka.common.protocol.ProtoUtils; import org.apache.kafka.common.protocol.types.Struct; import org.apache.kafka.common.record.MemoryRecords; -import org.apache.kafka.common.record.Record; import org.apache.kafka.common.utils.CollectionUtils; import org.apache.kafka.common.utils.Utils; @@ -130,10 +129,10 @@ public class ProduceRequest extends AbstractRequest { return null; Map<TopicPartition, ProduceResponse.PartitionResponse> responseMap = new HashMap<>(); + ProduceResponse.PartitionResponse partitionResponse = new ProduceResponse.PartitionResponse(Errors.forException(e)); - for (Map.Entry<TopicPartition, MemoryRecords> entry : partitionRecords.entrySet()) { - responseMap.put(entry.getKey(), new ProduceResponse.PartitionResponse(Errors.forException(e).code(), ProduceResponse.INVALID_OFFSET, Record.NO_TIMESTAMP)); - } + for (Map.Entry<TopicPartition, MemoryRecords> entry : partitionRecords.entrySet()) + responseMap.put(entry.getKey(), partitionResponse); short versionId = version(); switch (versionId) { http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java index 6c2125a..9eaaadf 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java @@ -14,9 +14,11 @@ package org.apache.kafka.common.requests; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.protocol.Errors; import org.apache.kafka.common.protocol.ProtoUtils; import org.apache.kafka.common.protocol.types.Schema; import org.apache.kafka.common.protocol.types.Struct; +import org.apache.kafka.common.record.Record; import org.apache.kafka.common.utils.CollectionUtils; import java.nio.ByteBuffer; @@ -61,7 +63,7 @@ public class ProduceResponse extends AbstractResponse { */ private static final String BASE_OFFSET_KEY_NAME = "base_offset"; - private static final String TIMESTAMP_KEY_NAME = "timestamp"; + private static final String LOG_APPEND_TIME_KEY_NAME = "log_append_time"; private final Map<TopicPartition, PartitionResponse> responses; private final int throttleTime; @@ -114,11 +116,11 @@ public class ProduceResponse extends AbstractResponse { for (Object partResponse : topicRespStruct.getArray(PARTITION_RESPONSES_KEY_NAME)) { Struct partRespStruct = (Struct) partResponse; int partition = partRespStruct.getInt(PARTITION_KEY_NAME); - short errorCode = partRespStruct.getShort(ERROR_CODE_KEY_NAME); + Errors error = Errors.forCode(partRespStruct.getShort(ERROR_CODE_KEY_NAME)); long offset = partRespStruct.getLong(BASE_OFFSET_KEY_NAME); - long timestamp = partRespStruct.getLong(TIMESTAMP_KEY_NAME); + long logAppendTime = partRespStruct.getLong(LOG_APPEND_TIME_KEY_NAME); TopicPartition tp = new TopicPartition(topic, partition); - responses.put(tp, new PartitionResponse(errorCode, offset, timestamp)); + responses.put(tp, new PartitionResponse(error, offset, logAppendTime)); } } this.throttleTime = struct.getInt(THROTTLE_TIME_KEY_NAME); @@ -135,10 +137,10 @@ public class ProduceResponse extends AbstractResponse { PartitionResponse part = partitionEntry.getValue(); Struct partStruct = topicData.instance(PARTITION_RESPONSES_KEY_NAME) .set(PARTITION_KEY_NAME, partitionEntry.getKey()) - .set(ERROR_CODE_KEY_NAME, part.errorCode) + .set(ERROR_CODE_KEY_NAME, part.error.code()) .set(BASE_OFFSET_KEY_NAME, part.baseOffset); - if (partStruct.hasField(TIMESTAMP_KEY_NAME)) - partStruct.set(TIMESTAMP_KEY_NAME, part.timestamp); + if (partStruct.hasField(LOG_APPEND_TIME_KEY_NAME)) + partStruct.set(LOG_APPEND_TIME_KEY_NAME, part.logAppendTime); partitionArray.add(partStruct); } topicData.set(PARTITION_RESPONSES_KEY_NAME, partitionArray.toArray()); @@ -156,14 +158,18 @@ public class ProduceResponse extends AbstractResponse { } public static final class PartitionResponse { - public short errorCode; + public Errors error; public long baseOffset; - public long timestamp; + public long logAppendTime; - public PartitionResponse(short errorCode, long baseOffset, long timestamp) { - this.errorCode = errorCode; + public PartitionResponse(Errors error) { + this(error, INVALID_OFFSET, Record.NO_TIMESTAMP); + } + + public PartitionResponse(Errors error, long baseOffset, long logAppendTime) { + this.error = error; this.baseOffset = baseOffset; - this.timestamp = timestamp; + this.logAppendTime = logAppendTime; } @Override @@ -171,11 +177,11 @@ public class ProduceResponse extends AbstractResponse { StringBuilder b = new StringBuilder(); b.append('{'); b.append("error: "); - b.append(errorCode); + b.append(error); b.append(",offset: "); b.append(baseOffset); - b.append(",timestamp: "); - b.append(timestamp); + b.append(",logAppendTime: "); + b.append(logAppendTime); b.append('}'); return b.toString(); } http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java ---------------------------------------------------------------------- diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java index 90256bb..13f7fda 100644 --- a/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java @@ -22,6 +22,7 @@ import org.apache.kafka.common.Cluster; import org.apache.kafka.common.KafkaException; import org.apache.kafka.common.Node; import org.apache.kafka.common.PartitionInfo; +import org.apache.kafka.common.config.ConfigException; import org.apache.kafka.common.network.Selectable; import org.apache.kafka.common.serialization.ByteArraySerializer; import org.apache.kafka.common.serialization.StringSerializer; @@ -53,6 +54,20 @@ import static org.junit.Assert.fail; public class KafkaProducerTest { @Test + public void testConstructorWithSerializers() { + Properties producerProps = new Properties(); + producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000"); + new KafkaProducer<>(producerProps, new ByteArraySerializer(), new ByteArraySerializer()).close(); + } + + @Test(expected = ConfigException.class) + public void testNoSerializerProvided() { + Properties producerProps = new Properties(); + producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000"); + new KafkaProducer(producerProps); + } + + @Test public void testConstructorFailureCloseResource() { Properties props = new Properties(); props.setProperty(ProducerConfig.CLIENT_ID_CONFIG, "testConstructorClose"); http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/clients/src/test/java/org/apache/kafka/clients/producer/RecordSendTest.java ---------------------------------------------------------------------- diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/RecordSendTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/RecordSendTest.java index d820dab..a420d61 100644 --- a/clients/src/test/java/org/apache/kafka/clients/producer/RecordSendTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/producer/RecordSendTest.java @@ -35,16 +35,16 @@ import org.junit.Test; public class RecordSendTest { - private TopicPartition topicPartition = new TopicPartition("test", 0); - private long baseOffset = 45; - private long relOffset = 5; + private final TopicPartition topicPartition = new TopicPartition("test", 0); + private final long baseOffset = 45; + private final long relOffset = 5; /** * Test that waiting on a request that never completes times out */ @Test public void testTimeout() throws Exception { - ProduceRequestResult request = new ProduceRequestResult(); + ProduceRequestResult request = new ProduceRequestResult(topicPartition); FutureRecordMetadata future = new FutureRecordMetadata(request, relOffset, Record.NO_TIMESTAMP, 0, 0, 0); assertFalse("Request is not completed", future.isDone()); @@ -54,7 +54,7 @@ public class RecordSendTest { } catch (TimeoutException e) { /* this is good */ } - request.done(topicPartition, baseOffset, null); + request.done(baseOffset, Record.NO_TIMESTAMP, null); assertTrue(future.isDone()); assertEquals(baseOffset + relOffset, future.get().offset()); } @@ -81,15 +81,13 @@ public class RecordSendTest { /* create a new request result that will be completed after the given timeout */ public ProduceRequestResult asyncRequest(final long baseOffset, final RuntimeException error, final long timeout) { - final ProduceRequestResult request = new ProduceRequestResult(); + final ProduceRequestResult request = new ProduceRequestResult(topicPartition); Thread thread = new Thread() { public void run() { try { sleep(timeout); - request.done(topicPartition, baseOffset, error); - } catch (InterruptedException e) { - e.printStackTrace(); - } + request.done(baseOffset, Record.NO_TIMESTAMP, error); + } catch (InterruptedException e) { } } }; thread.start(); http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java ---------------------------------------------------------------------- diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java index 00c536c..7f5fe15 100644 --- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java @@ -272,7 +272,8 @@ public class SenderTest { } private ProduceResponse produceResponse(TopicPartition tp, long offset, int error, int throttleTimeMs) { - ProduceResponse.PartitionResponse resp = new ProduceResponse.PartitionResponse((short) error, offset, Record.NO_TIMESTAMP); + ProduceResponse.PartitionResponse resp = new ProduceResponse.PartitionResponse(Errors.forCode((short) error), + offset, Record.NO_TIMESTAMP); Map<TopicPartition, ProduceResponse.PartitionResponse> partResp = Collections.singletonMap(tp, resp); return new ProduceResponse(partResp, throttleTimeMs); } http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java ---------------------------------------------------------------------- diff --git a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java index a5ed806..7b71f8f 100644 --- a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java +++ b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java @@ -183,7 +183,8 @@ public class RequestResponseTest { @Test public void produceResponseVersionTest() { Map<TopicPartition, ProduceResponse.PartitionResponse> responseData = new HashMap<>(); - responseData.put(new TopicPartition("test", 0), new ProduceResponse.PartitionResponse(Errors.NONE.code(), 10000, Record.NO_TIMESTAMP)); + responseData.put(new TopicPartition("test", 0), new ProduceResponse.PartitionResponse(Errors.NONE, + 10000, Record.NO_TIMESTAMP)); ProduceResponse v0Response = new ProduceResponse(responseData); ProduceResponse v1Response = new ProduceResponse(responseData, 10, 1); ProduceResponse v2Response = new ProduceResponse(responseData, 10, 2); @@ -455,7 +456,8 @@ public class RequestResponseTest { private ProduceResponse createProduceResponse() { Map<TopicPartition, ProduceResponse.PartitionResponse> responseData = new HashMap<>(); - responseData.put(new TopicPartition("test", 0), new ProduceResponse.PartitionResponse(Errors.NONE.code(), 10000, Record.NO_TIMESTAMP)); + responseData.put(new TopicPartition("test", 0), new ProduceResponse.PartitionResponse(Errors.NONE, + 10000, Record.NO_TIMESTAMP)); return new ProduceResponse(responseData, 0); } http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala b/core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala index 1163e72..459297e 100644 --- a/core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala +++ b/core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala @@ -160,16 +160,15 @@ class GroupMetadataManager(val brokerId: Int, // construct the error status in the propagated assignment response // in the cache val status = responseStatus(groupMetadataPartition) - val statusError = Errors.forCode(status.errorCode) - val responseError = if (statusError == Errors.NONE) { + val responseError = if (status.error == Errors.NONE) { Errors.NONE } else { debug(s"Metadata from group ${group.groupId} with generation $generationId failed when appending to log " + - s"due to ${statusError.exceptionName}") + s"due to ${status.error.exceptionName}") // transform the log append error code to the corresponding the commit status error code - statusError match { + status.error match { case Errors.UNKNOWN_TOPIC_OR_PARTITION | Errors.NOT_ENOUGH_REPLICAS | Errors.NOT_ENOUGH_REPLICAS_AFTER_APPEND => @@ -186,13 +185,13 @@ class GroupMetadataManager(val brokerId: Int, | Errors.INVALID_FETCH_SIZE => error(s"Appending metadata message for group ${group.groupId} generation $generationId failed due to " + - s"${statusError.exceptionName}, returning UNKNOWN error code to the client") + s"${status.error.exceptionName}, returning UNKNOWN error code to the client") Errors.UNKNOWN case other => error(s"Appending metadata message for group ${group.groupId} generation $generationId failed " + - s"due to unexpected error: ${statusError.exceptionName}") + s"due to unexpected error: ${status.error.exceptionName}") other } @@ -254,11 +253,10 @@ class GroupMetadataManager(val brokerId: Int, // construct the commit response status and insert // the offset and metadata to cache if the append status has no error val status = responseStatus(offsetTopicPartition) - val statusError = Errors.forCode(status.errorCode) val responseCode = group synchronized { - if (statusError == Errors.NONE) { + if (status.error == Errors.NONE) { if (!group.is(Dead)) { filteredOffsetMetadata.foreach { case (topicPartition, offsetAndMetadata) => group.completePendingOffsetWrite(topicPartition, offsetAndMetadata) @@ -273,10 +271,10 @@ class GroupMetadataManager(val brokerId: Int, } debug(s"Offset commit $filteredOffsetMetadata from group ${group.groupId}, consumer $consumerId " + - s"with generation $generationId failed when appending to log due to ${statusError.exceptionName}") + s"with generation $generationId failed when appending to log due to ${status.error.exceptionName}") // transform the log append error code to the corresponding the commit status error code - val responseError = statusError match { + val responseError = status.error match { case Errors.UNKNOWN_TOPIC_OR_PARTITION | Errors.NOT_ENOUGH_REPLICAS | Errors.NOT_ENOUGH_REPLICAS_AFTER_APPEND => http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/core/src/main/scala/kafka/server/DelayedFetch.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/server/DelayedFetch.scala b/core/src/main/scala/kafka/server/DelayedFetch.scala index 0a1884a..a05131a 100644 --- a/core/src/main/scala/kafka/server/DelayedFetch.scala +++ b/core/src/main/scala/kafka/server/DelayedFetch.scala @@ -150,7 +150,7 @@ class DelayedFetch(delayMs: Long, ) val fetchPartitionData = logReadResults.map { case (tp, result) => - tp -> FetchPartitionData(result.errorCode, result.hw, result.info.records) + tp -> FetchPartitionData(result.error, result.hw, result.info.records) } responseCallback(fetchPartitionData) http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/core/src/main/scala/kafka/server/DelayedProduce.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/server/DelayedProduce.scala b/core/src/main/scala/kafka/server/DelayedProduce.scala index 1af0bfb..f27dff3 100644 --- a/core/src/main/scala/kafka/server/DelayedProduce.scala +++ b/core/src/main/scala/kafka/server/DelayedProduce.scala @@ -33,7 +33,7 @@ case class ProducePartitionStatus(requiredOffset: Long, responseStatus: Partitio @volatile var acksPending = false override def toString = "[acksPending: %b, error: %d, startOffset: %d, requiredOffset: %d]" - .format(acksPending, responseStatus.errorCode, responseStatus.baseOffset, requiredOffset) + .format(acksPending, responseStatus.error.code, responseStatus.baseOffset, requiredOffset) } /** @@ -58,10 +58,10 @@ class DelayedProduce(delayMs: Long, // first update the acks pending variable according to the error code produceMetadata.produceStatus.foreach { case (topicPartition, status) => - if (status.responseStatus.errorCode == Errors.NONE.code) { + if (status.responseStatus.error == Errors.NONE) { // Timeout error state will be cleared when required acks are received status.acksPending = true - status.responseStatus.errorCode = Errors.REQUEST_TIMED_OUT.code + status.responseStatus.error = Errors.REQUEST_TIMED_OUT } else { status.acksPending = false } @@ -95,7 +95,7 @@ class DelayedProduce(delayMs: Long, // Case B.1 || B.2 if (error != Errors.NONE || hasEnough) { status.acksPending = false - status.responseStatus.errorCode = error.code + status.responseStatus.error = error } } } http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/core/src/main/scala/kafka/server/KafkaApis.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala index b99ce4d..3ce27db 100644 --- a/core/src/main/scala/kafka/server/KafkaApis.scala +++ b/core/src/main/scala/kafka/server/KafkaApis.scala @@ -359,21 +359,19 @@ class KafkaApis(val requestChannel: RequestChannel, def sendResponseCallback(responseStatus: Map[TopicPartition, PartitionResponse]) { val mergedResponseStatus = responseStatus ++ - unauthorizedForWriteRequestInfo.mapValues(_ => - new PartitionResponse(Errors.TOPIC_AUTHORIZATION_FAILED.code, -1, Record.NO_TIMESTAMP)) ++ - nonExistingOrUnauthorizedForDescribeTopics.mapValues(_ => - new PartitionResponse(Errors.UNKNOWN_TOPIC_OR_PARTITION.code, -1, Record.NO_TIMESTAMP)) + unauthorizedForWriteRequestInfo.mapValues(_ => new PartitionResponse(Errors.TOPIC_AUTHORIZATION_FAILED)) ++ + nonExistingOrUnauthorizedForDescribeTopics.mapValues(_ => new PartitionResponse(Errors.UNKNOWN_TOPIC_OR_PARTITION)) var errorInResponse = false mergedResponseStatus.foreach { case (topicPartition, status) => - if (status.errorCode != Errors.NONE.code) { + if (status.error != Errors.NONE) { errorInResponse = true debug("Produce request with correlation id %d from client %s on partition %s failed due to %s".format( request.header.correlationId, request.header.clientId, topicPartition, - Errors.forCode(status.errorCode).exceptionName)) + status.error.exceptionName)) } } @@ -384,7 +382,7 @@ class KafkaApis(val requestChannel: RequestChannel, // the producer client will know that some error has happened and will refresh its metadata if (errorInResponse) { val exceptionsSummary = mergedResponseStatus.map { case (topicPartition, status) => - topicPartition -> Errors.forCode(status.errorCode).exceptionName + topicPartition -> status.error.exceptionName }.mkString(", ") info( s"Closing connection due to error during produce request with correlation id ${request.header.correlationId} " + @@ -481,7 +479,7 @@ class KafkaApis(val requestChannel: RequestChannel, FetchPartitionData(data.error, data.hw, data.records.toMessageFormat(Record.MAGIC_VALUE_V0)) } else data - tp -> new FetchResponse.PartitionData(convertedData.error, convertedData.hw, convertedData.records) + tp -> new FetchResponse.PartitionData(convertedData.error.code, convertedData.hw, convertedData.records) } } http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/core/src/main/scala/kafka/server/ReplicaManager.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala index be2e04a..dd8fc03 100644 --- a/core/src/main/scala/kafka/server/ReplicaManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaManager.scala @@ -45,10 +45,10 @@ import scala.collection.JavaConverters._ /* * Result metadata of a log append operation on the log */ -case class LogAppendResult(info: LogAppendInfo, error: Option[Throwable] = None) { - def errorCode = error match { - case None => Errors.NONE.code - case Some(e) => Errors.forException(e).code +case class LogAppendResult(info: LogAppendInfo, exception: Option[Throwable] = None) { + def error: Errors = exception match { + case None => Errors.NONE + case Some(e) => Errors.forException(e) } } @@ -66,20 +66,19 @@ case class LogReadResult(info: FetchDataInfo, leaderLogEndOffset: Long, fetchTimeMs: Long, readSize: Int, - error: Option[Throwable] = None) { + exception: Option[Throwable] = None) { - def errorCode = error match { - case None => Errors.NONE.code - case Some(e) => Errors.forException(e).code + def error: Errors = exception match { + case None => Errors.NONE + case Some(e) => Errors.forException(e) } - override def toString = { - "Fetch Data: [%s], HW: [%d], leaderLogEndOffset: [%d], readSize: [%d], error: [%s]" - .format(info, hw, leaderLogEndOffset, readSize, error) - } + override def toString = + s"Fetch Data: [$info], HW: [$hw], leaderLogEndOffset: [$leaderLogEndOffset], readSize: [$readSize], error: [$error]" + } -case class FetchPartitionData(error: Short = Errors.NONE.code, hw: Long = -1L, records: Records) +case class FetchPartitionData(error: Errors = Errors.NONE, hw: Long = -1L, records: Records) object LogReadResult { val UnknownLogReadResult = LogReadResult(info = FetchDataInfo(LogOffsetMetadata.UnknownOffsetMetadata, MemoryRecords.EMPTY), @@ -315,7 +314,7 @@ class ReplicaManager(val config: KafkaConfig, topicPartition -> ProducePartitionStatus( result.info.lastOffset + 1, // required offset - new PartitionResponse(result.errorCode, result.info.firstOffset, result.info.logAppendTime)) // response status + new PartitionResponse(result.error, result.info.firstOffset, result.info.logAppendTime)) // response status } if (delayedRequestRequired(requiredAcks, entriesPerPartition, localProduceResults)) { @@ -340,7 +339,7 @@ class ReplicaManager(val config: KafkaConfig, // If required.acks is outside accepted range, something is wrong with the client // Just return an error and don't handle the request at all val responseStatus = entriesPerPartition.map { case (topicPartition, _) => - topicPartition -> new PartitionResponse(Errors.INVALID_REQUIRED_ACKS.code, + topicPartition -> new PartitionResponse(Errors.INVALID_REQUIRED_ACKS, LogAppendInfo.UnknownLogAppendInfo.firstOffset, Record.NO_TIMESTAMP) } responseCallback(responseStatus) @@ -357,7 +356,7 @@ class ReplicaManager(val config: KafkaConfig, localProduceResults: Map[TopicPartition, LogAppendResult]): Boolean = { requiredAcks == -1 && entriesPerPartition.nonEmpty && - localProduceResults.values.count(_.error.isDefined) < entriesPerPartition.size + localProduceResults.values.count(_.exception.isDefined) < entriesPerPartition.size } private def isValidRequiredAcks(requiredAcks: Short): Boolean = { @@ -464,7 +463,7 @@ class ReplicaManager(val config: KafkaConfig, val logReadResultValues = logReadResults.map { case (_, v) => v } val bytesReadable = logReadResultValues.map(_.info.records.sizeInBytes).sum val errorReadingData = logReadResultValues.foldLeft(false) ((errorIncurred, readResult) => - errorIncurred || (readResult.errorCode != Errors.NONE.code)) + errorIncurred || (readResult.error != Errors.NONE)) // respond immediately if 1) fetch request does not want to wait // 2) fetch request does not require any data @@ -472,7 +471,7 @@ class ReplicaManager(val config: KafkaConfig, // 4) some error happens while reading data if (timeout <= 0 || fetchInfos.isEmpty || bytesReadable >= fetchMinBytes || errorReadingData) { val fetchPartitionData = logReadResults.map { case (tp, result) => - tp -> FetchPartitionData(result.errorCode, result.hw, result.info.records) + tp -> FetchPartitionData(result.error, result.hw, result.info.records) } responseCallback(fetchPartitionData) } else { @@ -567,7 +566,7 @@ class ReplicaManager(val config: KafkaConfig, leaderLogEndOffset = initialLogEndOffset, fetchTimeMs = fetchTimeMs, readSize = partitionFetchSize, - error = None) + exception = None) } catch { // NOTE: Failed fetch requests metric is not incremented for known exceptions since it // is supposed to indicate un-expected failure of a broker in handling a fetch request @@ -580,7 +579,7 @@ class ReplicaManager(val config: KafkaConfig, leaderLogEndOffset = -1L, fetchTimeMs = -1L, readSize = partitionFetchSize, - error = Some(e)) + exception = Some(e)) case e: Throwable => BrokerTopicStats.getBrokerTopicStats(tp.topic).failedFetchRequestRate.mark() BrokerTopicStats.getBrokerAllTopicsStats().failedFetchRequestRate.mark() @@ -590,7 +589,7 @@ class ReplicaManager(val config: KafkaConfig, leaderLogEndOffset = -1L, fetchTimeMs = -1L, readSize = partitionFetchSize, - error = Some(e)) + exception = Some(e)) } } http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala index d43d1af..c9d35af 100644 --- a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala @@ -107,7 +107,7 @@ class AuthorizerIntegrationTest extends BaseRequestTest { val RequestKeyToErrorCode = Map[ApiKeys, (Nothing) => Short]( ApiKeys.METADATA -> ((resp: requests.MetadataResponse) => resp.errors().asScala.find(_._1 == topic).getOrElse(("test", Errors.NONE))._2.code), - ApiKeys.PRODUCE -> ((resp: requests.ProduceResponse) => resp.responses().asScala.find(_._1 == tp).get._2.errorCode), + ApiKeys.PRODUCE -> ((resp: requests.ProduceResponse) => resp.responses().asScala.find(_._1 == tp).get._2.error.code), ApiKeys.FETCH -> ((resp: requests.FetchResponse) => resp.responseData().asScala.find(_._1 == tp).get._2.errorCode), ApiKeys.LIST_OFFSETS -> ((resp: requests.ListOffsetResponse) => resp.responseData().asScala.find(_._1 == tp).get._2.errorCode), ApiKeys.OFFSET_COMMIT -> ((resp: requests.OffsetCommitResponse) => resp.responseData().asScala.find(_._1 == tp).get._2), http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala b/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala index 9ebc7e3..ad61a37 100644 --- a/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala +++ b/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala @@ -187,13 +187,13 @@ abstract class BaseProducerSendTest extends KafkaServerTestHarness { def onCompletion(metadata: RecordMetadata, exception: Exception) { if (exception == null) { - assertEquals(offset, metadata.offset()) - assertEquals(topic, metadata.topic()) + assertEquals(offset, metadata.offset) + assertEquals(topic, metadata.topic) if (timestampType == TimestampType.CREATE_TIME) - assertEquals(baseTimestamp + timestampDiff, metadata.timestamp()) + assertEquals(baseTimestamp + timestampDiff, metadata.timestamp) else - assertTrue(metadata.timestamp() >= startTime && metadata.timestamp() <= System.currentTimeMillis()) - assertEquals(partition, metadata.partition()) + assertTrue(metadata.timestamp >= startTime && metadata.timestamp <= System.currentTimeMillis()) + assertEquals(partition, metadata.partition) offset += 1 timestampDiff += 1 } else { @@ -211,11 +211,18 @@ abstract class BaseProducerSendTest extends KafkaServerTestHarness { topicProps.setProperty(LogConfig.MessageTimestampTypeProp, "CreateTime") TestUtils.createTopic(zkUtils, topic, 1, 2, servers, topicProps) - for (i <- 1 to numRecords) { - val record = new ProducerRecord[Array[Byte], Array[Byte]](topic, partition, baseTimestamp + i, "key".getBytes, "value".getBytes) - producer.send(record, callback) + val recordAndFutures = for (i <- 1 to numRecords) yield { + val record = new ProducerRecord(topic, partition, baseTimestamp + i, "key".getBytes, "value".getBytes) + (record, producer.send(record, callback)) } producer.close(20000L, TimeUnit.MILLISECONDS) + recordAndFutures.foreach { case (record, future) => + val recordMetadata = future.get + if (timestampType == TimestampType.LOG_APPEND_TIME) + assertTrue(recordMetadata.timestamp >= startTime && recordMetadata.timestamp <= System.currentTimeMillis()) + else + assertEquals(record.timestamp, recordMetadata.timestamp) + } assertEquals(s"Should have offset $numRecords but only successfully sent ${callback.offset}", numRecords, callback.offset) } finally { producer.close() http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/core/src/test/scala/integration/kafka/api/PlaintextProducerSendTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/integration/kafka/api/PlaintextProducerSendTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextProducerSendTest.scala index a75e7c7..956fe61 100644 --- a/core/src/test/scala/integration/kafka/api/PlaintextProducerSendTest.scala +++ b/core/src/test/scala/integration/kafka/api/PlaintextProducerSendTest.scala @@ -22,53 +22,23 @@ import java.util.concurrent.ExecutionException import kafka.log.LogConfig import kafka.utils.TestUtils - import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord} -import org.apache.kafka.common.config.ConfigException import org.apache.kafka.common.errors.{InvalidTimestampException, SerializationException} import org.apache.kafka.common.record.TimestampType -import org.apache.kafka.common.serialization.ByteArraySerializer import org.junit.Assert._ import org.junit.Test class PlaintextProducerSendTest extends BaseProducerSendTest { - @Test - def testSerializerConstructors() { - try { - createNewProducerWithNoSerializer(brokerList) - fail("Instantiating a producer without specifying a serializer should cause a ConfigException") - } catch { - case _ : ConfigException => // this is ok - } - - // create a producer with explicit serializers should succeed - createNewProducerWithExplicitSerializer(brokerList) - } - - private def createNewProducerWithNoSerializer(brokerList: String): KafkaProducer[Array[Byte], Array[Byte]] = { - val producerProps = new Properties() - producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList) - registerProducer(new KafkaProducer(producerProps)) - } - - private def createNewProducerWithExplicitSerializer(brokerList: String): KafkaProducer[Array[Byte], Array[Byte]] = { + @Test(expected = classOf[SerializationException]) + def testWrongSerializer() { val producerProps = new Properties() producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList) - registerProducer(new KafkaProducer(producerProps, new ByteArraySerializer, new ByteArraySerializer)) - } - - @Test - def testWrongSerializer() { - // send a record with a wrong type should receive a serialization exception - try { - val producer = createProducerWithWrongSerializer(brokerList) - val record5 = new ProducerRecord[Array[Byte], Array[Byte]](topic, new Integer(0), "key".getBytes, "value".getBytes) - producer.send(record5) - fail("Should have gotten a SerializationException") - } catch { - case _: SerializationException => // this is ok - } + producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer") + producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer") + val producer = registerProducer(new KafkaProducer(producerProps)) + val record = new ProducerRecord[Array[Byte], Array[Byte]](topic, new Integer(0), "key".getBytes, "value".getBytes) + producer.send(record) } @Test @@ -96,7 +66,7 @@ class PlaintextProducerSendTest extends BaseProducerSendTest { try { // Send a message to auto-create the topic - val record = new ProducerRecord[Array[Byte], Array[Byte]](topic, null, "key".getBytes, "value".getBytes) + val record = new ProducerRecord(topic, null, "key".getBytes, "value".getBytes) assertEquals("Should have offset 0", 0L, producer.send(record).get.offset) // double check that the topic is created with leader elected @@ -137,12 +107,4 @@ class PlaintextProducerSendTest extends BaseProducerSendTest { } } - private def createProducerWithWrongSerializer(brokerList: String): KafkaProducer[Array[Byte], Array[Byte]] = { - val producerProps = new Properties() - producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList) - producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer") - producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer") - registerProducer(new KafkaProducer(producerProps)) - } - } http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/core/src/test/scala/unit/kafka/coordinator/GroupCoordinatorResponseTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/unit/kafka/coordinator/GroupCoordinatorResponseTest.scala b/core/src/test/scala/unit/kafka/coordinator/GroupCoordinatorResponseTest.scala index d3de16d..11f9102 100644 --- a/core/src/test/scala/unit/kafka/coordinator/GroupCoordinatorResponseTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/GroupCoordinatorResponseTest.scala @@ -1052,7 +1052,7 @@ class GroupCoordinatorResponseTest extends JUnitSuite { EasyMock.capture(capturedArgument))).andAnswer(new IAnswer[Unit] { override def answer = capturedArgument.getValue.apply( Map(new TopicPartition(Topic.GroupMetadataTopicName, groupPartitionId) -> - new PartitionResponse(Errors.NONE.code, 0L, Record.NO_TIMESTAMP) + new PartitionResponse(Errors.NONE, 0L, Record.NO_TIMESTAMP) ) )}) EasyMock.expect(replicaManager.getMagicAndTimestampType(EasyMock.anyObject())) @@ -1134,7 +1134,7 @@ class GroupCoordinatorResponseTest extends JUnitSuite { EasyMock.capture(capturedArgument))).andAnswer(new IAnswer[Unit] { override def answer = capturedArgument.getValue.apply( Map(new TopicPartition(Topic.GroupMetadataTopicName, groupPartitionId) -> - new PartitionResponse(Errors.NONE.code, 0L, Record.NO_TIMESTAMP) + new PartitionResponse(Errors.NONE, 0L, Record.NO_TIMESTAMP) ) )}) EasyMock.expect(replicaManager.getMagicAndTimestampType(EasyMock.anyObject())) http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/core/src/test/scala/unit/kafka/coordinator/GroupMetadataManagerTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/unit/kafka/coordinator/GroupMetadataManagerTest.scala b/core/src/test/scala/unit/kafka/coordinator/GroupMetadataManagerTest.scala index e3dca32..9478122 100644 --- a/core/src/test/scala/unit/kafka/coordinator/GroupMetadataManagerTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/GroupMetadataManagerTest.scala @@ -613,7 +613,7 @@ class GroupMetadataManagerTest { EasyMock.capture(capturedArgument))).andAnswer(new IAnswer[Unit] { override def answer = capturedArgument.getValue.apply( Map(new TopicPartition(Topic.GroupMetadataTopicName, groupPartitionId) -> - new PartitionResponse(error.code, 0L, Record.NO_TIMESTAMP) + new PartitionResponse(error, 0L, Record.NO_TIMESTAMP) ) )}) EasyMock.expect(replicaManager.getMagicAndTimestampType(EasyMock.anyObject())) http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/core/src/test/scala/unit/kafka/server/EdgeCaseRequestTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/unit/kafka/server/EdgeCaseRequestTest.scala b/core/src/test/scala/unit/kafka/server/EdgeCaseRequestTest.scala index 5c53ffa..1f9e18b 100755 --- a/core/src/test/scala/unit/kafka/server/EdgeCaseRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/EdgeCaseRequestTest.scala @@ -27,7 +27,7 @@ import kafka.utils._ import org.apache.kafka.common.TopicPartition import org.apache.kafka.common.network.ListenerName import org.apache.kafka.common.protocol.types.Type -import org.apache.kafka.common.protocol.{ApiKeys, SecurityProtocol} +import org.apache.kafka.common.protocol.{ApiKeys, Errors, SecurityProtocol} import org.apache.kafka.common.record.MemoryRecords import org.apache.kafka.common.requests.{ProduceRequest, ProduceResponse, ResponseHeader} import org.junit.Assert._ @@ -138,7 +138,7 @@ class EdgeCaseRequestTest extends KafkaServerTestHarness { val partitionResponse = produceResponse.responses().get(topicPartition) assertNotNull(partitionResponse) - assertEquals("There should be no error", 0, partitionResponse.errorCode) + assertEquals("There should be no error", Errors.NONE, partitionResponse.error) } @Test http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/core/src/test/scala/unit/kafka/server/ProduceRequestTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/unit/kafka/server/ProduceRequestTest.scala b/core/src/test/scala/unit/kafka/server/ProduceRequestTest.scala index 8ed93d9..b05be9d 100644 --- a/core/src/test/scala/unit/kafka/server/ProduceRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/ProduceRequestTest.scala @@ -45,9 +45,9 @@ class ProduceRequestTest extends BaseRequestTest { assertEquals(1, produceResponse.responses.size) val (tp, partitionResponse) = produceResponse.responses.asScala.head assertEquals(topicPartition, tp) - assertEquals(Errors.NONE.code, partitionResponse.errorCode) + assertEquals(Errors.NONE, partitionResponse.error) assertEquals(expectedOffset, partitionResponse.baseOffset) - assertEquals(-1, partitionResponse.timestamp) + assertEquals(-1, partitionResponse.logAppendTime) partitionResponse } @@ -82,9 +82,9 @@ class ProduceRequestTest extends BaseRequestTest { assertEquals(1, produceResponse.responses.size) val (tp, partitionResponse) = produceResponse.responses.asScala.head assertEquals(topicPartition, tp) - assertEquals(Errors.CORRUPT_MESSAGE.code, partitionResponse.errorCode) + assertEquals(Errors.CORRUPT_MESSAGE, partitionResponse.error) assertEquals(-1, partitionResponse.baseOffset) - assertEquals(-1, partitionResponse.timestamp) + assertEquals(-1, partitionResponse.logAppendTime) } private def sendProduceRequest(leaderId: Int, request: ProduceRequest): ProduceResponse = { http://git-wip-us.apache.org/repos/asf/kafka/blob/8b3c6c0a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala index 00959f1..d5075aa 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala @@ -101,7 +101,7 @@ class ReplicaManagerTest { new AtomicBoolean(false), QuotaFactory.instantiate(config, metrics, time).follower, Option(this.getClass.getName)) try { def callback(responseStatus: Map[TopicPartition, PartitionResponse]) = { - assert(responseStatus.values.head.errorCode == Errors.INVALID_REQUIRED_ACKS.code) + assert(responseStatus.values.head.error == Errors.INVALID_REQUIRED_ACKS) } rm.appendRecords( timeout = 0, @@ -128,13 +128,15 @@ class ReplicaManagerTest { try { var produceCallbackFired = false def produceCallback(responseStatus: Map[TopicPartition, PartitionResponse]) = { - assertEquals("Should give NotLeaderForPartitionException", Errors.NOT_LEADER_FOR_PARTITION.code, responseStatus.values.head.errorCode) + assertEquals("Should give NotLeaderForPartitionException", Errors.NOT_LEADER_FOR_PARTITION, + responseStatus.values.head.error) produceCallbackFired = true } var fetchCallbackFired = false def fetchCallback(responseStatus: Seq[(TopicPartition, FetchPartitionData)]) = { - assertEquals("Should give NotLeaderForPartitionException", Errors.NOT_LEADER_FOR_PARTITION.code, responseStatus.map(_._2).head.error) + assertEquals("Should give NotLeaderForPartitionException", Errors.NOT_LEADER_FOR_PARTITION, + responseStatus.map(_._2).head.error) fetchCallbackFired = true } @@ -229,7 +231,7 @@ class ReplicaManagerTest { var fetchError = 0 var fetchedRecords: Records = null def fetchCallback(responseStatus: Seq[(TopicPartition, FetchPartitionData)]) = { - fetchError = responseStatus.map(_._2).head.error + fetchError = responseStatus.map(_._2).head.error.code fetchedRecords = responseStatus.map(_._2).head.records fetchCallbackFired = true }
