Skip to content

MINOR: hygene cleanup in TransactionManagerTest #5951

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

Merged
omkreddy merged 1 commit into from
Dec 3, 2018
Merged

MINOR: hygene cleanup in TransactionManagerTest #5951

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
177 changes: 75 additions & 102 deletions ...nts/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,6 @@
import org.apache.kafka.common.record.MutableRecordBatch;
import org.apache.kafka.common.record.Record;
import org.apache.kafka.common.record.RecordBatch;
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.AddOffsetsToTxnRequest;
import org.apache.kafka.common.requests.AddOffsetsToTxnResponse;
import org.apache.kafka.common.requests.AddPartitionsToTxnRequest;
Expand Down Expand Up @@ -634,7 +633,7 @@ public void testBasicTransaction() throws InterruptedException {
prepareFindCoordinatorResponse(Errors.NONE, false, CoordinatorType.GROUP, consumerGroupId);
prepareTxnOffsetCommitResponse(consumerGroupId, pid, epoch, txnOffsetCommitResponse);

assertEquals(null, transactionManager.coordinator(CoordinatorType.GROUP));
assertNull(transactionManager.coordinator(CoordinatorType.GROUP));
sender.run(time.milliseconds()); // try to send TxnOffsetCommitRequest, but find we don't have a group coordinator.
sender.run(time.milliseconds()); // send find coordinator for group request
assertNotNull(transactionManager.coordinator(CoordinatorType.GROUP));
Expand Down Expand Up @@ -671,14 +670,11 @@ public void testDisconnectAndRetry() {
@Test
public void testUnsupportedFindCoordinator() {
transactionManager.initializeTransactions();
client.prepareUnsupportedVersionResponse(new MockClient.RequestMatcher() {
@Override
public boolean matches(AbstractRequest body) {
FindCoordinatorRequest findCoordinatorRequest = (FindCoordinatorRequest) body;
assertEquals(findCoordinatorRequest.coordinatorType(), CoordinatorType.TRANSACTION);
assertEquals(findCoordinatorRequest.coordinatorKey(), transactionalId);
return true;
}
client.prepareUnsupportedVersionResponse(body -> {
FindCoordinatorRequest findCoordinatorRequest = (FindCoordinatorRequest) body;
assertEquals(findCoordinatorRequest.coordinatorType(), CoordinatorType.TRANSACTION);
assertEquals(findCoordinatorRequest.coordinatorKey(), transactionalId);
return true;
});

sender.run(time.milliseconds()); // InitProducerRequest is queued
Expand All @@ -697,14 +693,11 @@ public void testUnsupportedInitTransactions() {
assertFalse(transactionManager.hasError());
assertNotNull(transactionManager.coordinator(CoordinatorType.TRANSACTION));

client.prepareUnsupportedVersionResponse(new MockClient.RequestMatcher() {
@Override
public boolean matches(AbstractRequest body) {
InitProducerIdRequest initProducerIdRequest = (InitProducerIdRequest) body;
assertEquals(initProducerIdRequest.transactionalId(), transactionalId);
assertEquals(initProducerIdRequest.transactionTimeoutMs(), transactionTimeoutMs);
return true;
}
client.prepareUnsupportedVersionResponse(body -> {
InitProducerIdRequest initProducerIdRequest = (InitProducerIdRequest) body;
assertEquals(initProducerIdRequest.transactionalId(), transactionalId);
assertEquals(initProducerIdRequest.transactionTimeoutMs(), transactionTimeoutMs);
return true;
});

sender.run(time.milliseconds()); // InitProducerRequest is dequeued
Expand Down Expand Up @@ -759,7 +752,7 @@ public void testLookupCoordinatorOnDisconnectAfterSend() {
// FindCoordinator and InitPid requests.
sender.run(time.milliseconds());

assertEquals(null, transactionManager.coordinator(CoordinatorType.TRANSACTION));
assertNull(transactionManager.coordinator(CoordinatorType.TRANSACTION));
assertFalse(initPidResult.isCompleted());
assertFalse(transactionManager.hasProducerId());

Expand Down Expand Up @@ -795,7 +788,7 @@ public void testLookupCoordinatorOnDisconnectBeforeSend() {
sender.run(time.milliseconds());
time.sleep(110); // waiting for the blackout period for the node to expire.

assertEquals(null, transactionManager.coordinator(CoordinatorType.TRANSACTION));
assertNull(transactionManager.coordinator(CoordinatorType.TRANSACTION));
assertFalse(initPidResult.isCompleted());
assertFalse(transactionManager.hasProducerId());

Expand Down Expand Up @@ -827,7 +820,7 @@ public void testLookupCoordinatorOnNotCoordinatorError() {
prepareInitPidResponse(Errors.NOT_COORDINATOR, false, pid, epoch);
sender.run(time.milliseconds()); // send pid, get not coordinator. Should resend the FindCoordinator and InitPid requests

assertEquals(null, transactionManager.coordinator(CoordinatorType.TRANSACTION));
assertNull(transactionManager.coordinator(CoordinatorType.TRANSACTION));
assertFalse(initPidResult.isCompleted());
assertFalse(transactionManager.hasProducerId());

Expand Down Expand Up @@ -1402,6 +1395,7 @@ public void testDisallowCommitOnProduceFailure() throws InterruptedException {
commitResult.await();
fail(); // the get() must throw an exception.
} catch (KafkaException e) {
// Expected
}

try {
Expand Down Expand Up @@ -1835,7 +1829,7 @@ public void shouldNotSendAbortTxnRequestWhenOnlyAddPartitionsRequestFailed() {
}

@Test
public void shouldNotSendAbortTxnRequestWhenOnlyAddOffsetsRequestFailed() throws Exception {
public void shouldNotSendAbortTxnRequestWhenOnlyAddOffsetsRequestFailed() {
final long pid = 13131L;
final short epoch = 1;

Expand All @@ -1861,7 +1855,7 @@ public void shouldNotSendAbortTxnRequestWhenOnlyAddOffsetsRequestFailed() throws
}

@Test
public void shouldFailAbortIfAddOffsetsFailsWithFatalError() throws Exception {
public void shouldFailAbortIfAddOffsetsFailsWithFatalError() {
final long pid = 13131L;
final short epoch = 1;

Expand Down Expand Up @@ -1908,7 +1902,7 @@ public void testNoDrainWhenPartitionsPending() throws InterruptedException {
PartitionInfo part2 = new PartitionInfo(topic, 1, node2, null, null);

Cluster cluster = new Cluster(null, Arrays.asList(node1, node2), Arrays.asList(part1, part2),
Collections.<String>emptySet(), Collections.<String>emptySet());
Collections.emptySet(), Collections.emptySet());
Set<Node> nodes = new HashSet<>();
nodes.add(node1);
nodes.add(node2);
Expand Down Expand Up @@ -1945,8 +1939,8 @@ public void testAllowDrainInAbortableErrorState() throws InterruptedException {
// Try to drain a message destined for tp1, it should get drained.
Node node1 = new Node(1, "localhost", 1112);
PartitionInfo part1 = new PartitionInfo(topic, 1, node1, null, null);
Cluster cluster = new Cluster(null, Arrays.asList(node1), Arrays.asList(part1),
Collections.<String>emptySet(), Collections.<String>emptySet());
Cluster cluster = new Cluster(null, Collections.singletonList(node1), Collections.singletonList(part1),
Collections.emptySet(), Collections.emptySet());
accumulator.append(tp1, time.milliseconds(), "key".getBytes(),
"value".getBytes(), Record.EMPTY_HEADERS, null, MAX_BLOCK_TIMEOUT);
Map<Integer, List<ProducerBatch>> drainedBatches = accumulator.drain(cluster, Collections.singleton(node1),
Expand All @@ -1972,8 +1966,8 @@ public void testRaiseErrorWhenNoPartitionsPendingOnDrain() throws InterruptedExc
Node node1 = new Node(0, "localhost", 1111);
PartitionInfo part1 = new PartitionInfo(topic, 0, node1, null, null);

Cluster cluster = new Cluster(null, Arrays.asList(node1), Arrays.asList(part1),
Collections.<String>emptySet(), Collections.<String>emptySet());
Cluster cluster = new Cluster(null, Collections.singletonList(node1), Collections.singletonList(part1),
Collections.emptySet(), Collections.emptySet());
Set<Node> nodes = new HashSet<>();
nodes.add(node1);
Map<Integer, List<ProducerBatch>> drainedBatches = accumulator.drain(cluster, nodes, Integer.MAX_VALUE,
Expand Down Expand Up @@ -2012,7 +2006,7 @@ public void resendFailedProduceRequestAfterAbortableError() throws Exception {
}

@Test
public void testTransitionToAbortableErrorOnBatchExpiry() throws InterruptedException, ExecutionException {
public void testTransitionToAbortableErrorOnBatchExpiry() throws InterruptedException {
final long pid = 13131L;
final short epoch = 1;

Expand Down Expand Up @@ -2058,7 +2052,7 @@ public void testTransitionToAbortableErrorOnBatchExpiry() throws InterruptedExce
}

@Test
public void testTransitionToAbortableErrorOnMultipleBatchExpiry() throws InterruptedException, ExecutionException {
public void testTransitionToAbortableErrorOnMultipleBatchExpiry() throws InterruptedException {
final long pid = 13131L;
final short epoch = 1;

Expand Down Expand Up @@ -2123,7 +2117,7 @@ public void testTransitionToAbortableErrorOnMultipleBatchExpiry() throws Interru
}

@Test
public void testDropCommitOnBatchExpiry() throws InterruptedException, ExecutionException {
public void testDropCommitOnBatchExpiry() throws InterruptedException {
final long pid = 13131L;
final short epoch = 1;

Expand Down Expand Up @@ -2190,7 +2184,7 @@ public void testDropCommitOnBatchExpiry() throws InterruptedException, Execution
}

@Test
public void testTransitionToFatalErrorWhenRetriedBatchIsExpired() throws InterruptedException, ExecutionException {
public void testTransitionToFatalErrorWhenRetriedBatchIsExpired() throws InterruptedException {
final long pid = 13131L;
final short epoch = 1;

Expand Down Expand Up @@ -2249,7 +2243,7 @@ public void testTransitionToFatalErrorWhenRetriedBatchIsExpired() throws Interru
}

@Test
public void testShouldResetProducerStateAfterResolvingSequences() throws InterruptedException, ExecutionException {
public void testShouldResetProducerStateAfterResolvingSequences() {
// Create a TransactionManager without a transactionalId to test
// shouldResetProducerStateAfterResolvingSequences.
TransactionManager manager = new TransactionManager(logContext, null, transactionTimeoutMs,
Expand Down Expand Up @@ -2294,13 +2288,10 @@ private void verifyAddPartitionsFailsWithPartitionLevelError(final Errors error)
}

private void prepareAddPartitionsToTxn(final Map<TopicPartition, Errors> errors) {
client.prepareResponse(new MockClient.RequestMatcher() {
@Override
public boolean matches(AbstractRequest body) {
AddPartitionsToTxnRequest request = (AddPartitionsToTxnRequest) body;
assertEquals(new HashSet<>(request.partitions()), new HashSet<>(errors.keySet()));
return true;
}
client.prepareResponse(body -> {
AddPartitionsToTxnRequest request = (AddPartitionsToTxnRequest) body;
assertEquals(new HashSet<>(request.partitions()), new HashSet<>(errors.keySet()));
return true;
}, new AddPartitionsToTxnResponse(0, errors));
}

Expand All @@ -2311,26 +2302,20 @@ private void prepareAddPartitionsToTxn(final TopicPartition tp, final Errors err
private void prepareFindCoordinatorResponse(Errors error, boolean shouldDisconnect,
final CoordinatorType coordinatorType,
final String coordinatorKey) {
client.prepareResponse(new MockClient.RequestMatcher() {
@Override
public boolean matches(AbstractRequest body) {
FindCoordinatorRequest findCoordinatorRequest = (FindCoordinatorRequest) body;
assertEquals(findCoordinatorRequest.coordinatorType(), coordinatorType);
assertEquals(findCoordinatorRequest.coordinatorKey(), coordinatorKey);
return true;
}
client.prepareResponse(body -> {
FindCoordinatorRequest findCoordinatorRequest = (FindCoordinatorRequest) body;
assertEquals(findCoordinatorRequest.coordinatorType(), coordinatorType);
assertEquals(findCoordinatorRequest.coordinatorKey(), coordinatorKey);
return true;
}, new FindCoordinatorResponse(error, brokerNode), shouldDisconnect);
}

private void prepareInitPidResponse(Errors error, boolean shouldDisconnect, long pid, short epoch) {
client.prepareResponse(new MockClient.RequestMatcher() {
@Override
public boolean matches(AbstractRequest body) {
InitProducerIdRequest initProducerIdRequest = (InitProducerIdRequest) body;
assertEquals(initProducerIdRequest.transactionalId(), transactionalId);
assertEquals(initProducerIdRequest.transactionTimeoutMs(), transactionTimeoutMs);
return true;
}
client.prepareResponse(body -> {
InitProducerIdRequest initProducerIdRequest = (InitProducerIdRequest) body;
assertEquals(initProducerIdRequest.transactionalId(), transactionalId);
assertEquals(initProducerIdRequest.transactionTimeoutMs(), transactionTimeoutMs);
return true;
}, new InitProducerIdResponse(0, error, pid, epoch), shouldDisconnect);
}

Expand All @@ -2342,22 +2327,19 @@ private void prepareProduceResponse(Errors error, final long pid, final short ep
client.prepareResponse(produceRequestMatcher(pid, epoch), produceResponse(tp0, 0, error, 0));
}
private MockClient.RequestMatcher produceRequestMatcher(final long pid, final short epoch) {
return new MockClient.RequestMatcher() {
@Override
public boolean matches(AbstractRequest body) {
ProduceRequest produceRequest = (ProduceRequest) body;
MemoryRecords records = produceRequest.partitionRecordsOrFail().get(tp0);
assertNotNull(records);
Iterator<MutableRecordBatch> batchIterator = records.batches().iterator();
assertTrue(batchIterator.hasNext());
MutableRecordBatch batch = batchIterator.next();
assertFalse(batchIterator.hasNext());
assertTrue(batch.isTransactional());
assertEquals(pid, batch.producerId());
assertEquals(epoch, batch.producerEpoch());
assertEquals(transactionalId, produceRequest.transactionalId());
return true;
}
return body -> {
ProduceRequest produceRequest = (ProduceRequest) body;
MemoryRecords records = produceRequest.partitionRecordsOrFail().get(tp0);
assertNotNull(records);
Iterator<MutableRecordBatch> batchIterator = records.batches().iterator();
assertTrue(batchIterator.hasNext());
MutableRecordBatch batch = batchIterator.next();
assertFalse(batchIterator.hasNext());
assertTrue(batch.isTransactional());
assertEquals(pid, batch.producerId());
assertEquals(epoch, batch.producerEpoch());
assertEquals(transactionalId, produceRequest.transactionalId());
return true;
};
}

Expand All @@ -2375,16 +2357,13 @@ private void sendAddPartitionsToTxnResponse(Errors error, final TopicPartition t

private MockClient.RequestMatcher addPartitionsRequestMatcher(final TopicPartition topicPartition,
final short epoch, final long pid) {
return new MockClient.RequestMatcher() {
@Override
public boolean matches(AbstractRequest body) {
AddPartitionsToTxnRequest addPartitionsToTxnRequest = (AddPartitionsToTxnRequest) body;
assertEquals(pid, addPartitionsToTxnRequest.producerId());
assertEquals(epoch, addPartitionsToTxnRequest.producerEpoch());
assertEquals(singletonList(topicPartition), addPartitionsToTxnRequest.partitions());
assertEquals(transactionalId, addPartitionsToTxnRequest.transactionalId());
return true;
}
return body -> {
AddPartitionsToTxnRequest addPartitionsToTxnRequest = (AddPartitionsToTxnRequest) body;
assertEquals(pid, addPartitionsToTxnRequest.producerId());
assertEquals(epoch, addPartitionsToTxnRequest.producerEpoch());
assertEquals(singletonList(topicPartition), addPartitionsToTxnRequest.partitions());
assertEquals(transactionalId, addPartitionsToTxnRequest.transactionalId());
return true;
};
}

Expand All @@ -2397,33 +2376,27 @@ private void sendEndTxnResponse(Errors error, final TransactionResult result, fi
}

private MockClient.RequestMatcher endTxnMatcher(final TransactionResult result, final long pid, final short epoch) {
return new MockClient.RequestMatcher() {
@Override
public boolean matches(AbstractRequest body) {
EndTxnRequest endTxnRequest = (EndTxnRequest) body;
assertEquals(transactionalId, endTxnRequest.transactionalId());
assertEquals(pid, endTxnRequest.producerId());
assertEquals(epoch, endTxnRequest.producerEpoch());
assertEquals(result, endTxnRequest.command());
return true;
}
return body -> {
EndTxnRequest endTxnRequest = (EndTxnRequest) body;
assertEquals(transactionalId, endTxnRequest.transactionalId());
assertEquals(pid, endTxnRequest.producerId());
assertEquals(epoch, endTxnRequest.producerEpoch());
assertEquals(result, endTxnRequest.command());
return true;
};
}

private void prepareAddOffsetsToTxnResponse(final Errors error,
final String consumerGroupId,
final long producerId,
final short producerEpoch) {
client.prepareResponse(new MockClient.RequestMatcher() {
@Override
public boolean matches(AbstractRequest body) {
AddOffsetsToTxnRequest addOffsetsToTxnRequest = (AddOffsetsToTxnRequest) body;
assertEquals(consumerGroupId, addOffsetsToTxnRequest.consumerGroupId());
assertEquals(transactionalId, addOffsetsToTxnRequest.transactionalId());
assertEquals(producerId, addOffsetsToTxnRequest.producerId());
assertEquals(producerEpoch, addOffsetsToTxnRequest.producerEpoch());
return true;
}
client.prepareResponse(body -> {
AddOffsetsToTxnRequest addOffsetsToTxnRequest = (AddOffsetsToTxnRequest) body;
assertEquals(consumerGroupId, addOffsetsToTxnRequest.consumerGroupId());
assertEquals(transactionalId, addOffsetsToTxnRequest.transactionalId());
assertEquals(producerId, addOffsetsToTxnRequest.producerId());
assertEquals(producerEpoch, addOffsetsToTxnRequest.producerEpoch());
return true;
}, new AddOffsetsToTxnResponse(0, error));
}

Expand Down