Skip to content

KAFKA-2120 #320

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
MayureshGharat wants to merge 2 commits into from
Closed

KAFKA-2120 #320

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2d28c58
Trivial fix to get rid of unused statements in kafkaProducer
Oct 15, 2015
d9025a9
Addressed comments on the PR.
Oct 16, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 6 additions & 6 deletions clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -418,7 +418,7 @@ public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callbac
" to class " + producerConfig.getClass(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG).getName() +
" specified in key.serializer");
}
long remainingTime = checkMaybeGetRemainingTime(startTime);
checkMaybeGetRemainingTime(startTime);
byte[] serializedValue;
try {
serializedValue = valueSerializer.serialize(record.topic(), record.value());
Expand All @@ -427,14 +427,14 @@ public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callbac
" to class " + producerConfig.getClass(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG).getName() +
" specified in value.serializer");
}
remainingTime = checkMaybeGetRemainingTime(startTime);
checkMaybeGetRemainingTime(startTime);
int partition = partition(record, serializedKey, serializedValue, metadata.fetch());
remainingTime = checkMaybeGetRemainingTime(startTime);
checkMaybeGetRemainingTime(startTime);
int serializedSize = Records.LOG_OVERHEAD + Record.recordSize(serializedKey, serializedValue);
ensureValidRecordSize(serializedSize);
TopicPartition tp = new TopicPartition(record.topic(), partition);
log.trace("Sending record {} with callback {} to topic {} partition {}", record, callback, record.topic(), partition);
remainingTime = checkMaybeGetRemainingTime(startTime);
long remainingTime = checkMaybeGetRemainingTime(startTime);
RecordAccumulator.RecordAppendResult result = accumulator.append(tp, serializedKey, serializedValue, callback, remainingTime);
if (result.batchIsFull || result.newBatchCreated) {
log.trace("Waking up the sender since topic {} partition {} is either full or getting a new batch", record.topic(), partition);
Expand Down Expand Up @@ -679,7 +679,7 @@ private int partition(ProducerRecord<K, V> record, byte[] serializedKey , byte[]
}

/**
* Check and may be get the time elapsed since startTime.
* Check and maybe get the time elapsed since startTime.
* Throws a {@link org.apache.kafka.common.errors.TimeoutException} if the elapsed time
* is more than the max time to block (max.block.ms)
*
Expand All @@ -689,7 +689,7 @@ private int partition(ProducerRecord<K, V> record, byte[] serializedKey , byte[]
private long checkMaybeGetRemainingTime(long startTime) {
long elapsedTime = time.milliseconds() - startTime;
if (elapsedTime > maxBlockTimeMs) {
throw new TimeoutException("Request timed out");
throw new TimeoutException("Request timed out due to exceeding the maximum threshold of " + maxBlockTimeMs + " ms");
}
long remainingTime = maxBlockTimeMs - elapsedTime;

Expand Down