You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by jg...@apache.org on 2017/01/26 01:25:52 UTC
kafka git commit: KAFKA-4597;
RecordMetadata should return log append time when appropriate
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 <is...@juma.me.uk>
Reviewers: Ewen Cheslack-Postava <ew...@confluent.io>, Jason Gustafson <ja...@confluent.io>
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 <is...@juma.me.uk>
Authored: Wed Jan 25 17:25:10 2017 -0800
Committer: Jason Gustafson <ja...@confluent.io>
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
}