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 2019/01/22 17:35:04 UTC
[kafka] branch 0.11.0 updated: MINOR: Cleanup handling of mixed
transactional/idempotent records (#6172)
This is an automated email from the ASF dual-hosted git repository.
jgus pushed a commit to branch 0.11.0
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/0.11.0 by this push:
new 59cdb69 MINOR: Cleanup handling of mixed transactional/idempotent records (#6172)
59cdb69 is described below
commit 59cdb69983d9834b1e9225880248e37d57585481
Author: Jason Gustafson <ja...@confluent.io>
AuthorDate: Tue Jan 22 08:35:14 2019 -0800
MINOR: Cleanup handling of mixed transactional/idempotent records (#6172)
Reviewers: Rajini Sivaram <ra...@googlemail.com>, Ismael Juma <is...@juma.me.uk>, Colin Patrick McCabe <co...@cmccabe.xyz>
---
.../kafka/common/requests/ProduceRequest.java | 16 ++---
.../clients/producer/internals/SenderTest.java | 6 +-
.../kafka/common/requests/ProduceRequestTest.java | 76 ++++++++++++++++++----
core/src/main/scala/kafka/server/KafkaApis.scala | 8 ++-
4 files changed, 78 insertions(+), 28 deletions(-)
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 3d696c1..046d4df 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
@@ -109,8 +109,8 @@ public class ProduceRequest extends AbstractRequest {
// put in the purgatory (due to client throttling, it can take a while before the response is sent).
// Care should be taken in methods that use this field.
private volatile Map<TopicPartition, MemoryRecords> partitionRecords;
- private boolean transactional = false;
- private boolean idempotent = false;
+ private boolean hasTransactionalRecords = false;
+ private boolean hasIdempotentRecords = false;
private ProduceRequest(short version, short acks, int timeout, Map<TopicPartition, MemoryRecords> partitionRecords, String transactionalId) {
super(version);
@@ -167,8 +167,8 @@ public class ProduceRequest extends AbstractRequest {
if (iterator.hasNext())
throw new InvalidRecordException("Produce requests with version " + version + " are only allowed to " +
"contain exactly one record batch");
- idempotent = entry.hasProducerId();
- transactional = entry.isTransactional();
+ hasIdempotentRecords = hasIdempotentRecords || entry.hasProducerId();
+ hasTransactionalRecords = hasTransactionalRecords || entry.isTransactional();
}
// Note that we do not do similar validation for older versions to ensure compatibility with
@@ -269,12 +269,12 @@ public class ProduceRequest extends AbstractRequest {
return transactionalId;
}
- public boolean isTransactional() {
- return transactional;
+ public boolean hasTransactionalRecords() {
+ return hasTransactionalRecords;
}
- public boolean isIdempotent() {
- return idempotent;
+ public boolean hasIdempotentRecords() {
+ return hasIdempotentRecords;
}
/**
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 487aec8..892cc00 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
@@ -481,7 +481,7 @@ public class SenderTest {
client.prepareResponse(new MockClient.RequestMatcher() {
@Override
public boolean matches(AbstractRequest body) {
- return body instanceof ProduceRequest && ((ProduceRequest) body).isIdempotent();
+ return body instanceof ProduceRequest && ((ProduceRequest) body).hasIdempotentRecords();
}
}, produceResponse(tp0, -1, Errors.CLUSTER_AUTHORIZATION_FAILED, 0));
@@ -514,7 +514,7 @@ public class SenderTest {
client.prepareResponse(new MockClient.RequestMatcher() {
@Override
public boolean matches(AbstractRequest body) {
- return body instanceof ProduceRequest && ((ProduceRequest) body).isIdempotent();
+ return body instanceof ProduceRequest && ((ProduceRequest) body).hasIdempotentRecords();
}
}, produceResponse(tp0, -1, Errors.UNSUPPORTED_FOR_MESSAGE_FORMAT, 0));
@@ -546,7 +546,7 @@ public class SenderTest {
client.prepareUnsupportedVersionResponse(new MockClient.RequestMatcher() {
@Override
public boolean matches(AbstractRequest body) {
- return body instanceof ProduceRequest && ((ProduceRequest) body).isIdempotent();
+ return body instanceof ProduceRequest && ((ProduceRequest) body).hasIdempotentRecords();
}
});
diff --git a/clients/src/test/java/org/apache/kafka/common/requests/ProduceRequestTest.java b/clients/src/test/java/org/apache/kafka/common/requests/ProduceRequestTest.java
index 0e8f382..4ee6cc2 100644
--- a/clients/src/test/java/org/apache/kafka/common/requests/ProduceRequestTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/requests/ProduceRequestTest.java
@@ -25,6 +25,8 @@ import org.apache.kafka.common.record.SimpleRecord;
import org.junit.Test;
import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
@@ -49,38 +51,84 @@ public class ProduceRequestTest {
10,
Collections.singletonMap(
new TopicPartition("topic", 1), memoryRecords)).build();
- assertTrue(request.isTransactional());
+ assertTrue(request.hasTransactionalRecords());
}
@Test
public void shouldNotBeFlaggedAsTransactionalWhenNoRecords() throws Exception {
final ProduceRequest request = createNonIdempotentNonTransactionalRecords();
- assertFalse(request.isTransactional());
+ assertFalse(request.hasTransactionalRecords());
}
@Test
public void shouldNotBeFlaggedAsIdempotentWhenRecordsNotIdempotent() throws Exception {
final ProduceRequest request = createNonIdempotentNonTransactionalRecords();
- assertFalse(request.isTransactional());
+ assertFalse(request.hasTransactionalRecords());
}
@Test
public void shouldBeFlaggedAsIdempotentWhenIdempotentRecords() throws Exception {
final MemoryRecords memoryRecords = MemoryRecords.withIdempotentRecords(1,
- CompressionType.NONE,
- 1L,
- (short) 1,
- 1,
- 1,
- simpleRecord);
+ CompressionType.NONE,
+ 1L,
+ (short) 1,
+ 1,
+ 1,
+ simpleRecord);
final ProduceRequest request = new ProduceRequest.Builder(RecordBatch.CURRENT_MAGIC_VALUE,
- (short) -1,
- 10,
- Collections.singletonMap(
- new TopicPartition("topic", 1), memoryRecords)).build();
- assertTrue(request.isIdempotent());
+ (short) -1,
+ 10,
+ Collections.singletonMap(
+ new TopicPartition("topic", 1), memoryRecords)).build();
+ assertTrue(request.hasIdempotentRecords());
+ }
+
+ @Test
+ public void testMixedTransactionalData() {
+ final long producerId = 15L;
+ final short producerEpoch = 5;
+ final int sequence = 10;
+ final String transactionalId = "txnlId";
+
+ final MemoryRecords nonTxnRecords = MemoryRecords.withRecords(CompressionType.NONE,
+ new SimpleRecord("foo".getBytes()));
+ final MemoryRecords txnRecords = MemoryRecords.withTransactionalRecords(CompressionType.NONE, producerId,
+ producerEpoch, sequence, new SimpleRecord("bar".getBytes()));
+
+ final Map<TopicPartition, MemoryRecords> recordsByPartition = new LinkedHashMap<>();
+ recordsByPartition.put(new TopicPartition("foo", 0), txnRecords);
+ recordsByPartition.put(new TopicPartition("foo", 1), nonTxnRecords);
+
+ final ProduceRequest.Builder builder = new ProduceRequest.Builder(RecordBatch.CURRENT_MAGIC_VALUE, (short) -1, 5000,
+ recordsByPartition, transactionalId);
+
+ final ProduceRequest request = builder.build();
+ assertTrue(request.hasTransactionalRecords());
+ assertTrue(request.hasIdempotentRecords());
+ }
+
+ @Test
+ public void testMixedIdempotentData() {
+ final long producerId = 15L;
+ final short producerEpoch = 5;
+ final int sequence = 10;
+
+ final MemoryRecords nonTxnRecords = MemoryRecords.withRecords(CompressionType.NONE,
+ new SimpleRecord("foo".getBytes()));
+ final MemoryRecords txnRecords = MemoryRecords.withIdempotentRecords(CompressionType.NONE, producerId,
+ producerEpoch, sequence, new SimpleRecord("bar".getBytes()));
+
+ final Map<TopicPartition, MemoryRecords> recordsByPartition = new LinkedHashMap<>();
+ recordsByPartition.put(new TopicPartition("foo", 0), txnRecords);
+ recordsByPartition.put(new TopicPartition("foo", 1), nonTxnRecords);
+
+ final ProduceRequest.Builder builder = new ProduceRequest.Builder(RecordBatch.CURRENT_MAGIC_VALUE, (short) -1, 5000,
+ recordsByPartition, null);
+ final ProduceRequest request = builder.build();
+ assertFalse(request.hasTransactionalRecords());
+ assertTrue(request.hasIdempotentRecords());
}
private ProduceRequest createNonIdempotentNonTransactionalRecords() {
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala
index f600cb8..80b1fa3 100644
--- a/core/src/main/scala/kafka/server/KafkaApis.scala
+++ b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -378,14 +378,16 @@ class KafkaApis(val requestChannel: RequestChannel,
produceRequest.getErrorResponse(requestThrottleMs, error.exception))
}
- if (produceRequest.isTransactional) {
- if (!authorize(request.session, Write, new Resource(TransactionalId, produceRequest.transactionalId))) {
+ if (produceRequest.hasTransactionalRecords) {
+ val isAuthorizedTransactional = produceRequest.transactionalId != null &&
+ authorize(request.session, Write, Resource(TransactionalId, produceRequest.transactionalId))
+ if (!isAuthorizedTransactional) {
sendErrorResponse(Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED)
return
}
// Note that authorization to a transactionalId implies ProducerId authorization
- } else if (produceRequest.isIdempotent && !authorize(request.session, IdempotentWrite, Resource.ClusterResource)) {
+ } else if (produceRequest.hasIdempotentRecords && !authorize(request.session, IdempotentWrite, Resource.ClusterResource)) {
sendErrorResponse(Errors.CLUSTER_AUTHORIZATION_FAILED)
return
}