MINOR: Code Cleanup Clean up includes:
- Switching try-catch-finally blocks to try-with-resources when possible - Removing some seemingly unnecessary `SuppressWarnings` annotations - Resolving some Java warnings - Closing unclosed Closable objects - Removing unused code Author: Vahid Hashemian <vahidhashem...@us.ibm.com> Reviewers: Balint Molnar <balintmolna...@gmail.com>, Guozhang Wang <wangg...@gmail.com>, Matthias J. Sax <matth...@confluent.io>, Ismael Juma <ism...@juma.me.uk>, Jason Gustafson <ja...@confluent.io> Closes #3222 from vahidhashemian/minor/code_cleanup_1706 Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/f87d58b7 Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/f87d58b7 Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/f87d58b7 Branch: refs/heads/trunk Commit: f87d58b796977fdaefb089d17cb30b2071cd4485 Parents: 3bfc073 Author: Vahid Hashemian <vahidhashem...@us.ibm.com> Authored: Wed Jul 19 10:51:28 2017 -0700 Committer: Jason Gustafson <ja...@confluent.io> Committed: Wed Jul 19 10:51:28 2017 -0700 ---------------------------------------------------------------------- .../kafka/clients/producer/KafkaProducer.java | 3 +- .../producer/internals/TransactionManager.java | 1 - .../requests/OffsetsForLeaderEpochRequest.java | 4 +- .../clients/consumer/KafkaConsumerTest.java | 17 +- .../clients/producer/KafkaProducerTest.java | 33 ++-- .../clients/producer/MockProducerTest.java | 71 ++++++- .../clients/producer/internals/SenderTest.java | 9 +- .../utils/ByteBufferOutputStreamTest.java | 10 +- .../connect/file/FileStreamSourceTaskTest.java | 1 + .../connect/runtime/isolation/Plugins.java | 1 - .../connect/storage/FileOffsetBackingStore.java | 4 +- .../runtime/WorkerSinkTaskThreadedTest.java | 1 - .../storage/KafkaConfigBackingStoreTest.java | 17 +- .../kafka/connect/util/TopicAdminTest.java | 1 - .../kafka/connect/transforms/CastTest.java | 144 ++++++-------- .../connect/transforms/ExtractFieldTest.java | 9 +- .../kafka/connect/transforms/FlattenTest.java | 54 +++--- .../connect/transforms/HoistFieldTest.java | 9 +- .../connect/transforms/InsertFieldTest.java | 14 +- .../connect/transforms/RegexRouterTest.java | 5 +- .../connect/transforms/ReplaceFieldTest.java | 11 +- .../transforms/SetSchemaMetadataTest.java | 11 +- .../transforms/TimestampConverterTest.java | 128 ++++++------- .../connect/transforms/TimestampRouterTest.java | 8 +- .../connect/transforms/ValueToKeyTest.java | 9 +- .../apache/kafka/streams/kstream/KStream.java | 1 - .../kafka/streams/kstream/KStreamBuilder.java | 6 +- .../kstream/internals/KGroupedStreamImpl.java | 7 - .../kstream/internals/KTableKTableJoin.java | 1 - .../kstream/internals/KTableKTableLeftJoin.java | 1 - .../internals/KTableKTableOuterJoin.java | 1 - .../internals/KTableKTableRightJoin.java | 1 - .../internals/GlobalStateUpdateTask.java | 1 - .../processor/internals/StreamThread.java | 1 - .../internals/assignment/AssignmentInfo.java | 4 +- .../apache/kafka/streams/state/StateSerdes.java | 1 - .../state/internals/CachingKeyValueStore.java | 1 - .../state/internals/CachingWindowStore.java | 1 - .../ChangeLoggingSegmentedBytesStore.java | 1 - .../streams/state/internals/RocksDBStore.java | 1 - .../streams/kstream/KStreamBuilderTest.java | 10 +- ...reamSessionWindowAggregateProcessorTest.java | 2 - .../WindowedStreamPartitionerTest.java | 6 + .../streams/processor/TopologyBuilderTest.java | 14 +- .../processor/internals/QuickUnionTest.java | 1 - .../processor/internals/StandbyTaskTest.java | 2 - .../internals/StreamPartitionAssignorTest.java | 1 - .../processor/internals/StreamTaskTest.java | 1 - .../streams/state/KeyValueStoreTestDriver.java | 1 - .../ChangeLoggingSegmentedBytesStoreTest.java | 3 - .../CompositeReadOnlyKeyValueStoreTest.java | 4 - .../DelegatingPeekingKeyValueIteratorTest.java | 5 + ...gedSortedCacheKeyValueStoreIteratorTest.java | 2 + ...rtedCacheWrappedWindowStoreIteratorTest.java | 3 + .../MeteredSegmentedBytesStoreTest.java | 1 - .../RocksDBKeyValueStoreSupplierTest.java | 2 - .../RocksDBSessionStoreSupplierTest.java | 3 - .../RocksDBWindowStoreSupplierTest.java | 3 - .../state/internals/StoreChangeLoggerTest.java | 2 - .../streams/tests/BrokerCompatibilityTest.java | 1 + .../kafka/streams/tests/SmokeTestDriver.java | 1 - .../kafka/tools/ClientCompatibilityTest.java | 187 ++++++++++--------- 62 files changed, 419 insertions(+), 440 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kafka/blob/f87d58b7/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java index c54b739..5f854d1 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java @@ -302,7 +302,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> { keySerializer, valueSerializer); } - @SuppressWarnings({"unchecked", "deprecation"}) + @SuppressWarnings("unchecked") private KafkaProducer(ProducerConfig config, Serializer<K> keySerializer, Serializer<V> valueSerializer) { try { log.trace("Starting the Kafka producer"); @@ -339,7 +339,6 @@ public class KafkaProducer<K, V> implements Producer<K, V> { config.ignore(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG); this.valueSerializer = ensureExtended(valueSerializer); } - // load interceptors and make sure they get clientId userProvidedConfigs.put(ProducerConfig.CLIENT_ID_CONFIG, clientId); http://git-wip-us.apache.org/repos/asf/kafka/blob/f87d58b7/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java index 8cee794..c5542e4 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java @@ -628,7 +628,6 @@ public class TransactionManager { } @Override - @SuppressWarnings("unchecked") public void onComplete(ClientResponse response) { if (response.requestHeader().correlationId() != inFlightRequestCorrelationId) { fatalError(new RuntimeException("Detected more than one in-flight transactional request.")); http://git-wip-us.apache.org/repos/asf/kafka/blob/f87d58b7/clients/src/main/java/org/apache/kafka/common/requests/OffsetsForLeaderEpochRequest.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/requests/OffsetsForLeaderEpochRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/OffsetsForLeaderEpochRequest.java index f898a75..fc31d75 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/OffsetsForLeaderEpochRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/OffsetsForLeaderEpochRequest.java @@ -42,7 +42,7 @@ public class OffsetsForLeaderEpochRequest extends AbstractRequest { } public static class Builder extends AbstractRequest.Builder<OffsetsForLeaderEpochRequest> { - private Map<TopicPartition, Integer> epochsByPartition = new HashMap(); + private Map<TopicPartition, Integer> epochsByPartition = new HashMap<>(); public Builder() { super(ApiKeys.OFFSET_FOR_LEADER_EPOCH); @@ -129,7 +129,7 @@ public class OffsetsForLeaderEpochRequest extends AbstractRequest { @Override public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) { Errors error = Errors.forException(e); - Map<TopicPartition, EpochEndOffset> errorResponse = new HashMap(); + Map<TopicPartition, EpochEndOffset> errorResponse = new HashMap<>(); for (TopicPartition tp : epochsByPartition.keySet()) { errorResponse.put(tp, new EpochEndOffset(error, EpochEndOffset.UNDEFINED_EPOCH_OFFSET)); } http://git-wip-us.apache.org/repos/asf/kafka/blob/f87d58b7/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java ---------------------------------------------------------------------- diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java index 219c3f6..9fd7e19 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java @@ -129,13 +129,12 @@ public class KafkaConsumerTest { final int oldCloseCount = MockMetricsReporter.CLOSE_COUNT.get(); try { new KafkaConsumer<>(props, new ByteArrayDeserializer(), new ByteArrayDeserializer()); + Assert.fail("should have caught an exception and returned"); } catch (KafkaException e) { assertEquals(oldInitCount + 1, MockMetricsReporter.INIT_COUNT.get()); assertEquals(oldCloseCount + 1, MockMetricsReporter.CLOSE_COUNT.get()); assertEquals("Failed to construct kafka consumer", e.getMessage()); - return; } - Assert.fail("should have caught an exception and returned"); } @Test @@ -1191,23 +1190,17 @@ public class KafkaConsumerTest { @Test(expected = IllegalStateException.class) public void testPollWithEmptySubscription() { - KafkaConsumer<byte[], byte[]> consumer = newConsumer(); - consumer.subscribe(Collections.<String>emptyList()); - try { + try (KafkaConsumer<byte[], byte[]> consumer = newConsumer()) { + consumer.subscribe(Collections.<String>emptyList()); consumer.poll(0); - } finally { - consumer.close(); } } @Test(expected = IllegalStateException.class) public void testPollWithEmptyUserAssignment() { - KafkaConsumer<byte[], byte[]> consumer = newConsumer(); - consumer.assign(Collections.<TopicPartition>emptySet()); - try { + try (KafkaConsumer<byte[], byte[]> consumer = newConsumer()) { + consumer.assign(Collections.<TopicPartition>emptySet()); consumer.poll(0); - } finally { - consumer.close(); } } http://git-wip-us.apache.org/repos/asf/kafka/blob/f87d58b7/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 56c1b18..dd62457 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 @@ -87,16 +87,13 @@ public class KafkaProducerTest { final int oldInitCount = MockMetricsReporter.INIT_COUNT.get(); final int oldCloseCount = MockMetricsReporter.CLOSE_COUNT.get(); - try { - KafkaProducer<byte[], byte[]> producer = new KafkaProducer<byte[], byte[]>( - props, new ByteArraySerializer(), new ByteArraySerializer()); + try (KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(props, new ByteArraySerializer(), new ByteArraySerializer())) { + fail("should have caught an exception and returned"); } catch (KafkaException e) { assertEquals(oldInitCount + 1, MockMetricsReporter.INIT_COUNT.get()); assertEquals(oldCloseCount + 1, MockMetricsReporter.CLOSE_COUNT.get()); assertEquals("Failed to construct kafka producer", e.getMessage()); - return; } - fail("should have caught an exception and returned"); } @Test @@ -172,9 +169,7 @@ public class KafkaProducerTest { config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999"); config.put(ProducerConfig.SEND_BUFFER_CONFIG, Selectable.USE_DEFAULT_BUFFER_SIZE); config.put(ProducerConfig.RECEIVE_BUFFER_CONFIG, Selectable.USE_DEFAULT_BUFFER_SIZE); - KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>( - config, new ByteArraySerializer(), new ByteArraySerializer()); - producer.close(); + new KafkaProducer<>(config, new ByteArraySerializer(), new ByteArraySerializer()).close(); } @Test(expected = KafkaException.class) @@ -355,7 +350,7 @@ public class KafkaProducerTest { } Assert.assertTrue("Topic should still exist in metadata", metadata.containsTopic(topic)); } - + @PrepareOnlyThisForTest(Metadata.class) @Test public void testHeaders() throws Exception { @@ -369,8 +364,6 @@ public class KafkaProducerTest { MemberModifier.field(KafkaProducer.class, "metadata").set(producer, metadata); String topic = "topic"; - Collection<Node> nodes = Collections.singletonList(new Node(0, "host1", 1000)); - final Cluster cluster = new Cluster( "dummy", Collections.singletonList(new Node(0, "host1", 1000)), @@ -395,9 +388,9 @@ public class KafkaProducerTest { //ensure headers can be mutated pre send. record.headers().add(new RecordHeader("test", "header2".getBytes())); - + producer.send(record, null); - + //ensure headers are closed and cannot be mutated post send try { record.headers().add(new RecordHeader("test", "test".getBytes())); @@ -405,7 +398,7 @@ public class KafkaProducerTest { } catch (IllegalStateException ise) { //expected } - + //ensure existing headers are not changed, and last header for key is still original value assertTrue(Arrays.equals(record.headers().lastHeader("test").value(), "header2".getBytes())); @@ -436,7 +429,7 @@ public class KafkaProducerTest { assertEquals(Sensor.RecordingLevel.DEBUG, producer.metrics.config().recordLevel()); } } - + @PrepareOnlyThisForTest(Metadata.class) @Test public void testInterceptorPartitionSetOnTooLargeRecord() throws Exception { @@ -445,7 +438,7 @@ public class KafkaProducerTest { props.setProperty(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, "1"); String topic = "topic"; ProducerRecord<String, String> record = new ProducerRecord<>(topic, "value"); - + KafkaProducer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer()); Metadata metadata = PowerMock.createNiceMock(Metadata.class); @@ -457,20 +450,18 @@ public class KafkaProducerTest { Collections.<String>emptySet(), Collections.<String>emptySet()); EasyMock.expect(metadata.fetch()).andReturn(cluster).once(); - + // Mock interceptors field ProducerInterceptors interceptors = PowerMock.createMock(ProducerInterceptors.class); EasyMock.expect(interceptors.onSend(record)).andReturn(record); interceptors.onSendError(EasyMock.eq(record), EasyMock.<TopicPartition>notNull(), EasyMock.<Exception>notNull()); EasyMock.expectLastCall(); MemberModifier.field(KafkaProducer.class, "interceptors").set(producer, interceptors); - + PowerMock.replay(metadata); EasyMock.replay(interceptors); producer.send(record); - + EasyMock.verify(interceptors); - } - } http://git-wip-us.apache.org/repos/asf/kafka/blob/f87d58b7/clients/src/test/java/org/apache/kafka/clients/producer/MockProducerTest.java ---------------------------------------------------------------------- diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/MockProducerTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/MockProducerTest.java index eeb9b5f..d343194 100644 --- a/clients/src/test/java/org/apache/kafka/clients/producer/MockProducerTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/producer/MockProducerTest.java @@ -25,6 +25,7 @@ import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.ProducerFencedException; import org.apache.kafka.common.serialization.StringSerializer; import org.apache.kafka.test.MockSerializer; +import org.junit.After; import org.junit.Test; import java.util.ArrayList; @@ -47,14 +48,23 @@ import static org.junit.Assert.fail; public class MockProducerTest { private final String topic = "topic"; - private final MockProducer<byte[], byte[]> producer = new MockProducer<>(true, new MockSerializer(), new MockSerializer()); + private MockProducer<byte[], byte[]> producer; private final ProducerRecord<byte[], byte[]> record1 = new ProducerRecord<>(topic, "key1".getBytes(), "value1".getBytes()); private final ProducerRecord<byte[], byte[]> record2 = new ProducerRecord<>(topic, "key2".getBytes(), "value2".getBytes()); + private void buildMockProducer(boolean autoComplete) { + this.producer = new MockProducer<>(autoComplete, new MockSerializer(), new MockSerializer()); + } + + @After + public void cleanup() { + if (this.producer != null && !this.producer.closed()) + this.producer.close(); + } @Test - @SuppressWarnings("unchecked") public void testAutoCompleteMock() throws Exception { + buildMockProducer(true); Future<RecordMetadata> metadata = producer.send(record1); assertTrue("Send should be immediately complete", metadata.isDone()); assertFalse("Send should be successful", isError(metadata)); @@ -77,11 +87,12 @@ public class MockProducerTest { assertEquals("Partition should be correct", 1, metadata.get().partition()); producer.clear(); assertEquals("Clear should erase our history", 0, producer.history().size()); + producer.close(); } @Test public void testManualCompletion() throws Exception { - MockProducer<byte[], byte[]> producer = new MockProducer<>(false, new MockSerializer(), new MockSerializer()); + buildMockProducer(false); Future<RecordMetadata> md1 = producer.send(record1); assertFalse("Send shouldn't have completed", md1.isDone()); Future<RecordMetadata> md2 = producer.send(record2); @@ -98,7 +109,7 @@ public class MockProducerTest { assertEquals(e, err.getCause()); } assertFalse("No more requests to complete", producer.completeNext()); - + Future<RecordMetadata> md3 = producer.send(record1); Future<RecordMetadata> md4 = producer.send(record2); assertTrue("Requests should not be completed.", !md3.isDone() && !md4.isDone()); @@ -108,12 +119,14 @@ public class MockProducerTest { @Test public void shouldInitTransactions() { + buildMockProducer(true); producer.initTransactions(); assertTrue(producer.transactionInitialized()); } @Test public void shouldThrowOnInitTransactionIfProducerAlreadyInitializedForTransactions() { + buildMockProducer(true); producer.initTransactions(); try { producer.initTransactions(); @@ -123,11 +136,13 @@ public class MockProducerTest { @Test(expected = IllegalStateException.class) public void shouldThrowOnBeginTransactionIfTransactionsNotInitialized() { + buildMockProducer(true); producer.beginTransaction(); } @Test public void shouldBeginTransactions() { + buildMockProducer(true); producer.initTransactions(); producer.beginTransaction(); assertTrue(producer.transactionInFlight()); @@ -135,11 +150,13 @@ public class MockProducerTest { @Test(expected = IllegalStateException.class) public void shouldThrowOnSendOffsetsToTransactionIfTransactionsNotInitialized() { + buildMockProducer(true); producer.sendOffsetsToTransaction(null, null); } @Test public void shouldThrowOnSendOffsetsToTransactionTransactionIfNoTransactionGotStarted() { + buildMockProducer(true); producer.initTransactions(); try { producer.sendOffsetsToTransaction(null, null); @@ -149,11 +166,13 @@ public class MockProducerTest { @Test(expected = IllegalStateException.class) public void shouldThrowOnCommitIfTransactionsNotInitialized() { + buildMockProducer(true); producer.commitTransaction(); } @Test public void shouldThrowOnCommitTransactionIfNoTransactionGotStarted() { + buildMockProducer(true); producer.initTransactions(); try { producer.commitTransaction(); @@ -163,6 +182,7 @@ public class MockProducerTest { @Test public void shouldCommitEmptyTransaction() { + buildMockProducer(true); producer.initTransactions(); producer.beginTransaction(); producer.commitTransaction(); @@ -173,6 +193,7 @@ public class MockProducerTest { @Test public void shouldCountCommittedTransaction() { + buildMockProducer(true); producer.initTransactions(); producer.beginTransaction(); @@ -183,6 +204,7 @@ public class MockProducerTest { @Test public void shouldNotCountAbortedTransaction() { + buildMockProducer(true); producer.initTransactions(); producer.beginTransaction(); @@ -195,11 +217,13 @@ public class MockProducerTest { @Test(expected = IllegalStateException.class) public void shouldThrowOnAbortIfTransactionsNotInitialized() { + buildMockProducer(true); producer.abortTransaction(); } @Test public void shouldThrowOnAbortTransactionIfNoTransactionGotStarted() { + buildMockProducer(true); producer.initTransactions(); try { producer.abortTransaction(); @@ -209,6 +233,7 @@ public class MockProducerTest { @Test public void shouldAbortEmptyTransaction() { + buildMockProducer(true); producer.initTransactions(); producer.beginTransaction(); producer.abortTransaction(); @@ -219,6 +244,7 @@ public class MockProducerTest { @Test public void shouldAbortInFlightTransactionOnClose() { + buildMockProducer(true); producer.initTransactions(); producer.beginTransaction(); producer.close(); @@ -229,11 +255,13 @@ public class MockProducerTest { @Test(expected = IllegalStateException.class) public void shouldThrowFenceProducerIfTransactionsNotInitialized() { + buildMockProducer(true); producer.fenceProducer(); } @Test public void shouldThrowOnBeginTransactionsIfProducerGotFenced() { + buildMockProducer(true); producer.initTransactions(); producer.fenceProducer(); try { @@ -244,6 +272,7 @@ public class MockProducerTest { @Test public void shouldThrowOnSendIfProducerGotFenced() { + buildMockProducer(true); producer.initTransactions(); producer.fenceProducer(); try { @@ -254,6 +283,7 @@ public class MockProducerTest { @Test public void shouldThrowOnFlushIfProducerGotFenced() { + buildMockProducer(true); producer.initTransactions(); producer.fenceProducer(); try { @@ -264,6 +294,7 @@ public class MockProducerTest { @Test public void shouldThrowOnSendOffsetsToTransactionIfProducerGotFenced() { + buildMockProducer(true); producer.initTransactions(); producer.fenceProducer(); try { @@ -274,6 +305,7 @@ public class MockProducerTest { @Test public void shouldThrowOnCommitTransactionIfProducerGotFenced() { + buildMockProducer(true); producer.initTransactions(); producer.fenceProducer(); try { @@ -284,6 +316,7 @@ public class MockProducerTest { @Test public void shouldThrowOnAbortTransactionIfProducerGotFenced() { + buildMockProducer(true); producer.initTransactions(); producer.fenceProducer(); try { @@ -294,6 +327,7 @@ public class MockProducerTest { @Test public void shouldPublishMessagesOnlyAfterCommitIfTransactionsAreEnabled() { + buildMockProducer(true); producer.initTransactions(); producer.beginTransaction(); @@ -313,7 +347,7 @@ public class MockProducerTest { @Test public void shouldFlushOnCommitForNonAutoCompleteIfTransactionsAreEnabled() { - MockProducer<byte[], byte[]> producer = new MockProducer<>(false, new MockSerializer(), new MockSerializer()); + buildMockProducer(false); producer.initTransactions(); producer.beginTransaction(); @@ -331,6 +365,7 @@ public class MockProducerTest { @Test public void shouldDropMessagesOnAbortIfTransactionsAreEnabled() { + buildMockProducer(true); producer.initTransactions(); producer.beginTransaction(); @@ -346,7 +381,7 @@ public class MockProducerTest { @Test public void shouldThrowOnAbortForNonAutoCompleteIfTransactionsAreEnabled() throws Exception { - MockProducer<byte[], byte[]> producer = new MockProducer<>(false, new MockSerializer(), new MockSerializer()); + buildMockProducer(false); producer.initTransactions(); producer.beginTransaction(); @@ -359,6 +394,7 @@ public class MockProducerTest { @Test public void shouldPreserveCommittedMessagesOnAbortIfTransactionsAreEnabled() { + buildMockProducer(true); producer.initTransactions(); producer.beginTransaction(); @@ -378,6 +414,7 @@ public class MockProducerTest { @Test public void shouldPublishConsumerGroupOffsetsOnlyAfterCommitIfTransactionsAreEnabled() { + buildMockProducer(true); producer.initTransactions(); producer.beginTransaction(); @@ -410,6 +447,7 @@ public class MockProducerTest { @Test public void shouldThrowOnNullConsumerGroupIdWhenSendOffsetsToTransaction() { + buildMockProducer(true); producer.initTransactions(); producer.beginTransaction(); @@ -421,6 +459,7 @@ public class MockProducerTest { @Test public void shouldIgnoreEmptyOffsetsWhenSendOffsetsToTransaction() { + buildMockProducer(true); producer.initTransactions(); producer.beginTransaction(); producer.sendOffsetsToTransaction(Collections.<TopicPartition, OffsetAndMetadata>emptyMap(), "groupId"); @@ -429,6 +468,7 @@ public class MockProducerTest { @Test public void shouldAddOffsetsWhenSendOffsetsToTransaction() { + buildMockProducer(true); producer.initTransactions(); producer.beginTransaction(); @@ -445,6 +485,7 @@ public class MockProducerTest { @Test public void shouldResetSentOffsetsFlagOnlyWhenBeginningNewTransaction() { + buildMockProducer(true); producer.initTransactions(); producer.beginTransaction(); @@ -465,6 +506,7 @@ public class MockProducerTest { @Test public void shouldPublishLatestAndCumulativeConsumerGroupOffsetsOnlyAfterCommitIfTransactionsAreEnabled() { + buildMockProducer(true); producer.initTransactions(); producer.beginTransaction(); @@ -501,6 +543,7 @@ public class MockProducerTest { @Test public void shouldDropConsumerGroupOffsetsOnAbortIfTransactionsAreEnabled() { + buildMockProducer(true); producer.initTransactions(); producer.beginTransaction(); @@ -521,6 +564,7 @@ public class MockProducerTest { @Test public void shouldPreserveCommittedConsumerGroupsOffsetsOnAbortIfTransactionsAreEnabled() { + buildMockProducer(true); producer.initTransactions(); producer.beginTransaction(); @@ -545,6 +589,7 @@ public class MockProducerTest { @Test public void shouldThrowOnInitTransactionIfProducerIsClosed() { + buildMockProducer(true); producer.close(); try { producer.initTransactions(); @@ -554,6 +599,7 @@ public class MockProducerTest { @Test public void shouldThrowOnSendIfProducerIsClosed() { + buildMockProducer(true); producer.close(); try { producer.send(null); @@ -563,6 +609,7 @@ public class MockProducerTest { @Test public void shouldThrowOnBeginTransactionIfProducerIsClosed() { + buildMockProducer(true); producer.close(); try { producer.beginTransaction(); @@ -572,6 +619,7 @@ public class MockProducerTest { @Test public void shouldThrowSendOffsetsToTransactionIfProducerIsClosed() { + buildMockProducer(true); producer.close(); try { producer.sendOffsetsToTransaction(null, null); @@ -581,6 +629,7 @@ public class MockProducerTest { @Test public void shouldThrowOnCommitTransactionIfProducerIsClosed() { + buildMockProducer(true); producer.close(); try { producer.commitTransaction(); @@ -590,6 +639,7 @@ public class MockProducerTest { @Test public void shouldThrowOnAbortTransactionIfProducerIsClosed() { + buildMockProducer(true); producer.close(); try { producer.abortTransaction(); @@ -599,6 +649,7 @@ public class MockProducerTest { @Test public void shouldThrowOnCloseIfProducerIsClosed() { + buildMockProducer(true); producer.close(); try { producer.close(); @@ -608,6 +659,7 @@ public class MockProducerTest { @Test public void shouldThrowOnFenceProducerIfProducerIsClosed() { + buildMockProducer(true); producer.close(); try { producer.fenceProducer(); @@ -617,6 +669,7 @@ public class MockProducerTest { @Test public void shouldThrowOnFlushProducerIfProducerIsClosed() { + buildMockProducer(true); producer.close(); try { producer.flush(); @@ -626,25 +679,27 @@ public class MockProducerTest { @Test public void shouldBeFlushedIfNoBufferedRecords() { + buildMockProducer(true); assertTrue(producer.flushed()); } @Test public void shouldBeFlushedWithAutoCompleteIfBufferedRecords() { + buildMockProducer(true); producer.send(record1); assertTrue(producer.flushed()); } @Test public void shouldNotBeFlushedWithNoAutoCompleteIfBufferedRecords() { - MockProducer<byte[], byte[]> producer = new MockProducer<>(false, new MockSerializer(), new MockSerializer()); + buildMockProducer(false); producer.send(record1); assertFalse(producer.flushed()); } @Test public void shouldNotBeFlushedAfterFlush() { - MockProducer<byte[], byte[]> producer = new MockProducer<>(false, new MockSerializer(), new MockSerializer()); + buildMockProducer(false); producer.send(record1); producer.flush(); assertTrue(producer.flushed()); http://git-wip-us.apache.org/repos/asf/kafka/blob/f87d58b7/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 0537a35..d587de4 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 @@ -633,10 +633,9 @@ public class SenderTest { String topic = tp.topic(); // Set a good compression ratio. CompressionRatioEstimator.setEstimation(topic, CompressionType.GZIP, 0.2f); - Metrics m = new Metrics(); - accumulator = new RecordAccumulator(batchSize, 1024 * 1024, CompressionType.GZIP, 0L, 0L, m, time, - new ApiVersions(), txnManager); - try { + try (Metrics m = new Metrics()) { + accumulator = new RecordAccumulator(batchSize, 1024 * 1024, CompressionType.GZIP, 0L, 0L, m, time, + new ApiVersions(), txnManager); Sender sender = new Sender(client, metadata, this.accumulator, true, MAX_REQUEST_SIZE, ACKS_ALL, maxRetries, m, time, REQUEST_TIMEOUT, 1000L, txnManager, new ApiVersions()); // Create a two broker cluster, with partition 0 on broker 0 and partition 1 on broker 1 @@ -701,8 +700,6 @@ public class SenderTest { assertTrue("There should be a split", m.metrics().get(m.metricName("batch-split-rate", "producer-metrics")).value() > 0); - } finally { - m.close(); } } http://git-wip-us.apache.org/repos/asf/kafka/blob/f87d58b7/clients/src/test/java/org/apache/kafka/common/utils/ByteBufferOutputStreamTest.java ---------------------------------------------------------------------- diff --git a/clients/src/test/java/org/apache/kafka/common/utils/ByteBufferOutputStreamTest.java b/clients/src/test/java/org/apache/kafka/common/utils/ByteBufferOutputStreamTest.java index fbac719..26ba3b8 100644 --- a/clients/src/test/java/org/apache/kafka/common/utils/ByteBufferOutputStreamTest.java +++ b/clients/src/test/java/org/apache/kafka/common/utils/ByteBufferOutputStreamTest.java @@ -18,6 +18,7 @@ package org.apache.kafka.common.utils; import org.junit.Test; +import java.io.IOException; import java.nio.ByteBuffer; import static org.junit.Assert.assertArrayEquals; @@ -49,6 +50,7 @@ public class ByteBufferOutputStreamTest { byte[] bytes = new byte[5]; buffer.get(bytes); assertArrayEquals("hello".getBytes(), bytes); + output.close(); } @Test @@ -75,19 +77,20 @@ public class ByteBufferOutputStreamTest { byte[] bytes = new byte[5]; buffer.get(bytes); assertArrayEquals("hello".getBytes(), bytes); + output.close(); } @Test - public void testWriteByteBuffer() { + public void testWriteByteBuffer() throws IOException { testWriteByteBuffer(ByteBuffer.allocate(16)); } @Test - public void testWriteDirectByteBuffer() { + public void testWriteDirectByteBuffer() throws IOException { testWriteByteBuffer(ByteBuffer.allocateDirect(16)); } - private void testWriteByteBuffer(ByteBuffer input) { + private void testWriteByteBuffer(ByteBuffer input) throws IOException { long value = 234239230L; input.putLong(value); input.flip(); @@ -97,6 +100,7 @@ public class ByteBufferOutputStreamTest { assertEquals(8, input.position()); assertEquals(8, output.position()); assertEquals(value, output.buffer().getLong(0)); + output.close(); } } http://git-wip-us.apache.org/repos/asf/kafka/blob/f87d58b7/connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceTaskTest.java ---------------------------------------------------------------------- diff --git a/connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceTaskTest.java b/connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceTaskTest.java index eb91dbd..03fb774 100644 --- a/connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceTaskTest.java +++ b/connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceTaskTest.java @@ -123,6 +123,7 @@ public class FileStreamSourceTaskTest { assertEquals(Collections.singletonMap(FileStreamSourceTask.FILENAME_FIELD, tempFile.getAbsolutePath()), records.get(0).sourcePartition()); assertEquals(Collections.singletonMap(FileStreamSourceTask.POSITION_FIELD, 48L), records.get(0).sourceOffset()); + os.close(); task.stop(); } http://git-wip-us.apache.org/repos/asf/kafka/blob/f87d58b7/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/Plugins.java ---------------------------------------------------------------------- diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/Plugins.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/Plugins.java index 654f485..f7a5553 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/Plugins.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/Plugins.java @@ -137,7 +137,6 @@ public class Plugins { return delegatingLoader.transformations(); } - @SuppressWarnings("unchecked") public Connector newConnector(String connectorClassOrAlias) { Class<? extends Connector> klass; try { http://git-wip-us.apache.org/repos/asf/kafka/blob/f87d58b7/connect/runtime/src/main/java/org/apache/kafka/connect/storage/FileOffsetBackingStore.java ---------------------------------------------------------------------- diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/FileOffsetBackingStore.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/FileOffsetBackingStore.java index 32f5a38..0547fe6 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/FileOffsetBackingStore.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/FileOffsetBackingStore.java @@ -69,8 +69,7 @@ public class FileOffsetBackingStore extends MemoryOffsetBackingStore { @SuppressWarnings("unchecked") private void load() { - try { - ObjectInputStream is = new ObjectInputStream(new FileInputStream(file)); + try (ObjectInputStream is = new ObjectInputStream(new FileInputStream(file))) { Object obj = is.readObject(); if (!(obj instanceof HashMap)) throw new ConnectException("Expected HashMap but found " + obj.getClass()); @@ -81,7 +80,6 @@ public class FileOffsetBackingStore extends MemoryOffsetBackingStore { ByteBuffer value = (mapEntry.getValue() != null) ? ByteBuffer.wrap(mapEntry.getValue()) : null; data.put(key, value); } - is.close(); } catch (FileNotFoundException | EOFException e) { // FileNotFoundException: Ignore, may be new. // EOFException: Ignore, this means the file was missing or corrupt http://git-wip-us.apache.org/repos/asf/kafka/blob/f87d58b7/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskThreadedTest.java ---------------------------------------------------------------------- diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskThreadedTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskThreadedTest.java index 29a6b52..6f77f65 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskThreadedTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskThreadedTest.java @@ -115,7 +115,6 @@ public class WorkerSinkTaskThreadedTest extends ThreadedTest { private long recordsReturned; - @SuppressWarnings("unchecked") @Override public void setup() { super.setup(); http://git-wip-us.apache.org/repos/asf/kafka/blob/f87d58b7/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaConfigBackingStoreTest.java ---------------------------------------------------------------------- diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaConfigBackingStoreTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaConfigBackingStoreTest.java index 07d192b..e9dd18e 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaConfigBackingStoreTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaConfigBackingStoreTest.java @@ -135,7 +135,6 @@ public class KafkaConfigBackingStoreTest { KafkaBasedLog<String, byte[]> storeLog; private KafkaConfigBackingStore configStorage; - private String internalTopic; private Capture<String> capturedTopic = EasyMock.newCapture(); private Capture<Map<String, Object>> capturedProducerProps = EasyMock.newCapture(); private Capture<Map<String, Object>> capturedConsumerProps = EasyMock.newCapture(); @@ -363,7 +362,7 @@ public class KafkaConfigBackingStoreTest { new ConsumerRecord<>(TOPIC, 0, 2, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, TASK_CONFIG_KEYS.get(1), CONFIGS_SERIALIZED.get(2)), new ConsumerRecord<>(TOPIC, 0, 3, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, TARGET_STATE_KEYS.get(0), CONFIGS_SERIALIZED.get(3)), new ConsumerRecord<>(TOPIC, 0, 4, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, COMMIT_TASKS_CONFIG_KEYS.get(0), CONFIGS_SERIALIZED.get(4))); - LinkedHashMap<byte[], Struct> deserialized = new LinkedHashMap(); + LinkedHashMap<byte[], Struct> deserialized = new LinkedHashMap<>(); deserialized.put(CONFIGS_SERIALIZED.get(0), CONNECTOR_CONFIG_STRUCTS.get(0)); deserialized.put(CONFIGS_SERIALIZED.get(1), TASK_CONFIG_STRUCTS.get(0)); deserialized.put(CONFIGS_SERIALIZED.get(2), TASK_CONFIG_STRUCTS.get(0)); @@ -403,7 +402,7 @@ public class KafkaConfigBackingStoreTest { new ConsumerRecord<>(TOPIC, 0, 1, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, TASK_CONFIG_KEYS.get(0), CONFIGS_SERIALIZED.get(1)), new ConsumerRecord<>(TOPIC, 0, 2, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, TASK_CONFIG_KEYS.get(1), CONFIGS_SERIALIZED.get(2)), new ConsumerRecord<>(TOPIC, 0, 3, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, COMMIT_TASKS_CONFIG_KEYS.get(0), CONFIGS_SERIALIZED.get(3))); - LinkedHashMap<byte[], Struct> deserialized = new LinkedHashMap(); + LinkedHashMap<byte[], Struct> deserialized = new LinkedHashMap<>(); deserialized.put(CONFIGS_SERIALIZED.get(0), CONNECTOR_CONFIG_STRUCTS.get(0)); deserialized.put(CONFIGS_SERIALIZED.get(1), TASK_CONFIG_STRUCTS.get(0)); deserialized.put(CONFIGS_SERIALIZED.get(2), TASK_CONFIG_STRUCTS.get(0)); @@ -445,7 +444,7 @@ public class KafkaConfigBackingStoreTest { new ConsumerRecord<>(TOPIC, 0, 1, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, TASK_CONFIG_KEYS.get(0), CONFIGS_SERIALIZED.get(1)), new ConsumerRecord<>(TOPIC, 0, 2, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, TASK_CONFIG_KEYS.get(1), CONFIGS_SERIALIZED.get(2)), new ConsumerRecord<>(TOPIC, 0, 3, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, COMMIT_TASKS_CONFIG_KEYS.get(0), CONFIGS_SERIALIZED.get(3))); - LinkedHashMap<byte[], Struct> deserialized = new LinkedHashMap(); + LinkedHashMap<byte[], Struct> deserialized = new LinkedHashMap<>(); deserialized.put(CONFIGS_SERIALIZED.get(0), CONNECTOR_CONFIG_STRUCTS.get(0)); deserialized.put(CONFIGS_SERIALIZED.get(1), TASK_CONFIG_STRUCTS.get(0)); deserialized.put(CONFIGS_SERIALIZED.get(2), TASK_CONFIG_STRUCTS.get(0)); @@ -494,7 +493,7 @@ public class KafkaConfigBackingStoreTest { new ConsumerRecord<>(TOPIC, 0, 2, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, TASK_CONFIG_KEYS.get(1), CONFIGS_SERIALIZED.get(2)), new ConsumerRecord<>(TOPIC, 0, 3, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, TARGET_STATE_KEYS.get(0), CONFIGS_SERIALIZED.get(3)), new ConsumerRecord<>(TOPIC, 0, 4, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, COMMIT_TASKS_CONFIG_KEYS.get(0), CONFIGS_SERIALIZED.get(4))); - LinkedHashMap<byte[], Struct> deserialized = new LinkedHashMap(); + LinkedHashMap<byte[], Struct> deserialized = new LinkedHashMap<>(); deserialized.put(CONFIGS_SERIALIZED.get(0), CONNECTOR_CONFIG_STRUCTS.get(0)); deserialized.put(CONFIGS_SERIALIZED.get(1), TASK_CONFIG_STRUCTS.get(0)); deserialized.put(CONFIGS_SERIALIZED.get(2), TASK_CONFIG_STRUCTS.get(0)); @@ -540,7 +539,7 @@ public class KafkaConfigBackingStoreTest { // Connector after root update should make it through, task update shouldn't new ConsumerRecord<>(TOPIC, 0, 5, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, CONNECTOR_CONFIG_KEYS.get(0), CONFIGS_SERIALIZED.get(5)), new ConsumerRecord<>(TOPIC, 0, 6, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, TASK_CONFIG_KEYS.get(0), CONFIGS_SERIALIZED.get(6))); - LinkedHashMap<byte[], Struct> deserialized = new LinkedHashMap(); + LinkedHashMap<byte[], Struct> deserialized = new LinkedHashMap<>(); deserialized.put(CONFIGS_SERIALIZED.get(0), CONNECTOR_CONFIG_STRUCTS.get(0)); deserialized.put(CONFIGS_SERIALIZED.get(1), TASK_CONFIG_STRUCTS.get(0)); deserialized.put(CONFIGS_SERIALIZED.get(2), TASK_CONFIG_STRUCTS.get(0)); @@ -594,7 +593,7 @@ public class KafkaConfigBackingStoreTest { new ConsumerRecord<>(TOPIC, 0, 4, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, TARGET_STATE_KEYS.get(0), CONFIGS_SERIALIZED.get(4)), new ConsumerRecord<>(TOPIC, 0, 5, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, COMMIT_TASKS_CONFIG_KEYS.get(0), CONFIGS_SERIALIZED.get(5))); - LinkedHashMap<byte[], Struct> deserialized = new LinkedHashMap(); + LinkedHashMap<byte[], Struct> deserialized = new LinkedHashMap<>(); deserialized.put(CONFIGS_SERIALIZED.get(0), CONNECTOR_CONFIG_STRUCTS.get(0)); deserialized.put(CONFIGS_SERIALIZED.get(1), TASK_CONFIG_STRUCTS.get(0)); deserialized.put(CONFIGS_SERIALIZED.get(2), TASK_CONFIG_STRUCTS.get(0)); @@ -640,7 +639,7 @@ public class KafkaConfigBackingStoreTest { new ConsumerRecord<>(TOPIC, 0, 5, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, CONNECTOR_CONFIG_KEYS.get(0), CONFIGS_SERIALIZED.get(5)), new ConsumerRecord<>(TOPIC, 0, 6, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, TASK_CONFIG_KEYS.get(0), CONFIGS_SERIALIZED.get(6)), new ConsumerRecord<>(TOPIC, 0, 7, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, COMMIT_TASKS_CONFIG_KEYS.get(0), CONFIGS_SERIALIZED.get(7))); - LinkedHashMap<byte[], Struct> deserialized = new LinkedHashMap(); + LinkedHashMap<byte[], Struct> deserialized = new LinkedHashMap<>(); deserialized.put(CONFIGS_SERIALIZED.get(0), CONNECTOR_CONFIG_STRUCTS.get(0)); deserialized.put(CONFIGS_SERIALIZED.get(1), TASK_CONFIG_STRUCTS.get(0)); deserialized.put(CONFIGS_SERIALIZED.get(2), TASK_CONFIG_STRUCTS.get(0)); @@ -691,7 +690,7 @@ public class KafkaConfigBackingStoreTest { new ConsumerRecord<>(TOPIC, 0, 2, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, TASK_CONFIG_KEYS.get(1), CONFIGS_SERIALIZED.get(2)), new ConsumerRecord<>(TOPIC, 0, 4, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, COMMIT_TASKS_CONFIG_KEYS.get(0), CONFIGS_SERIALIZED.get(4)), new ConsumerRecord<>(TOPIC, 0, 5, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, TASK_CONFIG_KEYS.get(0), CONFIGS_SERIALIZED.get(5))); - LinkedHashMap<byte[], Struct> deserialized = new LinkedHashMap(); + LinkedHashMap<byte[], Struct> deserialized = new LinkedHashMap<>(); deserialized.put(CONFIGS_SERIALIZED.get(0), CONNECTOR_CONFIG_STRUCTS.get(0)); deserialized.put(CONFIGS_SERIALIZED.get(2), TASK_CONFIG_STRUCTS.get(0)); deserialized.put(CONFIGS_SERIALIZED.get(4), TASKS_COMMIT_STRUCT_TWO_TASK_CONNECTOR); http://git-wip-us.apache.org/repos/asf/kafka/blob/f87d58b7/connect/runtime/src/test/java/org/apache/kafka/connect/util/TopicAdminTest.java ---------------------------------------------------------------------- diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/util/TopicAdminTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/util/TopicAdminTest.java index bafbce8..f90b77f 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/util/TopicAdminTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/util/TopicAdminTest.java @@ -49,7 +49,6 @@ public class TopicAdminTest { @Test public void returnNullWithApiVersionMismatch() { final NewTopic newTopic = TopicAdmin.defineTopic("myTopic").partitions(1).compacted().build(); - boolean internal = false; Cluster cluster = createCluster(1); try (MockKafkaAdminClientEnv env = new MockKafkaAdminClientEnv(cluster)) { env.kafkaClient().setNode(cluster.controller()); http://git-wip-us.apache.org/repos/asf/kafka/blob/f87d58b7/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/CastTest.java ---------------------------------------------------------------------- diff --git a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/CastTest.java b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/CastTest.java index 88afafc..b190189 100644 --- a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/CastTest.java +++ b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/CastTest.java @@ -23,6 +23,7 @@ import org.apache.kafka.connect.data.SchemaBuilder; import org.apache.kafka.connect.data.Struct; import org.apache.kafka.connect.errors.DataException; import org.apache.kafka.connect.source.SourceRecord; +import org.junit.After; import org.junit.Test; import java.util.Collections; @@ -34,42 +35,44 @@ import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; public class CastTest { + private final Cast<SourceRecord> xformKey = new Cast.Key<>(); + private final Cast<SourceRecord> xformValue = new Cast.Value<>(); + + @After + public void teardown() { + xformKey.close(); + xformValue.close(); + } @Test(expected = ConfigException.class) public void testConfigEmpty() { - final Cast<SourceRecord> xform = new Cast.Key<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "")); + xformKey.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "")); } @Test(expected = ConfigException.class) public void testConfigInvalidSchemaType() { - final Cast<SourceRecord> xform = new Cast.Key<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "foo:faketype")); + xformKey.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "foo:faketype")); } @Test(expected = ConfigException.class) public void testConfigInvalidTargetType() { - final Cast<SourceRecord> xform = new Cast.Key<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "foo:array")); + xformKey.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "foo:array")); } @Test(expected = ConfigException.class) public void testConfigInvalidMap() { - final Cast<SourceRecord> xform = new Cast.Key<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "foo:int8:extra")); + xformKey.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "foo:int8:extra")); } @Test(expected = ConfigException.class) public void testConfigMixWholeAndFieldTransformation() { - final Cast<SourceRecord> xform = new Cast.Key<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "foo:int8,int32")); + xformKey.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "foo:int8,int32")); } @Test public void castWholeRecordKeyWithSchema() { - final Cast<SourceRecord> xform = new Cast.Key<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int8")); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, "topic", 0, + xformKey.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int8")); + SourceRecord transformed = xformKey.apply(new SourceRecord(null, null, "topic", 0, Schema.INT32_SCHEMA, 42, Schema.STRING_SCHEMA, "bogus")); assertEquals(Schema.Type.INT8, transformed.keySchema().type()); @@ -78,9 +81,8 @@ public class CastTest { @Test public void castWholeRecordValueWithSchemaInt8() { - final Cast<SourceRecord> xform = new Cast.Value<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int8")); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, "topic", 0, + xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int8")); + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, Schema.INT32_SCHEMA, 42)); assertEquals(Schema.Type.INT8, transformed.valueSchema().type()); @@ -89,9 +91,8 @@ public class CastTest { @Test public void castWholeRecordValueWithSchemaInt16() { - final Cast<SourceRecord> xform = new Cast.Value<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int16")); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, "topic", 0, + xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int16")); + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, Schema.INT32_SCHEMA, 42)); assertEquals(Schema.Type.INT16, transformed.valueSchema().type()); @@ -100,9 +101,8 @@ public class CastTest { @Test public void castWholeRecordValueWithSchemaInt32() { - final Cast<SourceRecord> xform = new Cast.Value<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int32")); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, "topic", 0, + xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int32")); + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, Schema.INT32_SCHEMA, 42)); assertEquals(Schema.Type.INT32, transformed.valueSchema().type()); @@ -111,9 +111,8 @@ public class CastTest { @Test public void castWholeRecordValueWithSchemaInt64() { - final Cast<SourceRecord> xform = new Cast.Value<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int64")); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, "topic", 0, + xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int64")); + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, Schema.INT32_SCHEMA, 42)); assertEquals(Schema.Type.INT64, transformed.valueSchema().type()); @@ -122,9 +121,8 @@ public class CastTest { @Test public void castWholeRecordValueWithSchemaFloat32() { - final Cast<SourceRecord> xform = new Cast.Value<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "float32")); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, "topic", 0, + xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "float32")); + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, Schema.INT32_SCHEMA, 42)); assertEquals(Schema.Type.FLOAT32, transformed.valueSchema().type()); @@ -133,9 +131,8 @@ public class CastTest { @Test public void castWholeRecordValueWithSchemaFloat64() { - final Cast<SourceRecord> xform = new Cast.Value<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "float64")); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, "topic", 0, + xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "float64")); + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, Schema.INT32_SCHEMA, 42)); assertEquals(Schema.Type.FLOAT64, transformed.valueSchema().type()); @@ -144,9 +141,8 @@ public class CastTest { @Test public void castWholeRecordValueWithSchemaBooleanTrue() { - final Cast<SourceRecord> xform = new Cast.Value<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "boolean")); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, "topic", 0, + xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "boolean")); + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, Schema.INT32_SCHEMA, 42)); assertEquals(Schema.Type.BOOLEAN, transformed.valueSchema().type()); @@ -155,9 +151,8 @@ public class CastTest { @Test public void castWholeRecordValueWithSchemaBooleanFalse() { - final Cast<SourceRecord> xform = new Cast.Value<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "boolean")); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, "topic", 0, + xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "boolean")); + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, Schema.INT32_SCHEMA, 0)); assertEquals(Schema.Type.BOOLEAN, transformed.valueSchema().type()); @@ -166,9 +161,8 @@ public class CastTest { @Test public void castWholeRecordValueWithSchemaString() { - final Cast<SourceRecord> xform = new Cast.Value<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "string")); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, "topic", 0, + xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "string")); + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, Schema.INT32_SCHEMA, 42)); assertEquals(Schema.Type.STRING, transformed.valueSchema().type()); @@ -178,9 +172,8 @@ public class CastTest { @Test public void castWholeRecordDefaultValue() { // Validate default value in schema is correctly converted - final Cast<SourceRecord> xform = new Cast.Value<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int32")); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, "topic", 0, + xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int32")); + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, SchemaBuilder.float32().defaultValue(-42.125f).build(), 42.125f)); assertEquals(Schema.Type.INT32, transformed.valueSchema().type()); @@ -190,9 +183,8 @@ public class CastTest { @Test public void castWholeRecordKeySchemaless() { - final Cast<SourceRecord> xform = new Cast.Key<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int8")); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, "topic", 0, + xformKey.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int8")); + SourceRecord transformed = xformKey.apply(new SourceRecord(null, null, "topic", 0, null, 42, Schema.STRING_SCHEMA, "bogus")); assertNull(transformed.keySchema()); @@ -201,9 +193,8 @@ public class CastTest { @Test public void castWholeRecordValueSchemalessInt8() { - final Cast<SourceRecord> xform = new Cast.Value<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int8")); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, "topic", 0, + xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int8")); + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, null, 42)); assertNull(transformed.valueSchema()); @@ -212,9 +203,8 @@ public class CastTest { @Test public void castWholeRecordValueSchemalessInt16() { - final Cast<SourceRecord> xform = new Cast.Value<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int16")); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, "topic", 0, + xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int16")); + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, null, 42)); assertNull(transformed.valueSchema()); @@ -223,9 +213,8 @@ public class CastTest { @Test public void castWholeRecordValueSchemalessInt32() { - final Cast<SourceRecord> xform = new Cast.Value<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int32")); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, "topic", 0, + xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int32")); + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, null, 42)); assertNull(transformed.valueSchema()); @@ -234,9 +223,8 @@ public class CastTest { @Test public void castWholeRecordValueSchemalessInt64() { - final Cast<SourceRecord> xform = new Cast.Value<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int64")); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, "topic", 0, + xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int64")); + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, null, 42)); assertNull(transformed.valueSchema()); @@ -245,9 +233,8 @@ public class CastTest { @Test public void castWholeRecordValueSchemalessFloat32() { - final Cast<SourceRecord> xform = new Cast.Value<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "float32")); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, "topic", 0, + xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "float32")); + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, null, 42)); assertNull(transformed.valueSchema()); @@ -256,9 +243,8 @@ public class CastTest { @Test public void castWholeRecordValueSchemalessFloat64() { - final Cast<SourceRecord> xform = new Cast.Value<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "float64")); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, "topic", 0, + xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "float64")); + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, null, 42)); assertNull(transformed.valueSchema()); @@ -267,9 +253,8 @@ public class CastTest { @Test public void castWholeRecordValueSchemalessBooleanTrue() { - final Cast<SourceRecord> xform = new Cast.Value<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "boolean")); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, "topic", 0, + xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "boolean")); + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, null, 42)); assertNull(transformed.valueSchema()); @@ -278,9 +263,8 @@ public class CastTest { @Test public void castWholeRecordValueSchemalessBooleanFalse() { - final Cast<SourceRecord> xform = new Cast.Value<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "boolean")); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, "topic", 0, + xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "boolean")); + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, null, 0)); assertNull(transformed.valueSchema()); @@ -289,9 +273,8 @@ public class CastTest { @Test public void castWholeRecordValueSchemalessString() { - final Cast<SourceRecord> xform = new Cast.Value<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "string")); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, "topic", 0, + xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "string")); + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, null, 42)); assertNull(transformed.valueSchema()); @@ -300,16 +283,14 @@ public class CastTest { @Test(expected = DataException.class) public void castWholeRecordValueSchemalessUnsupportedType() { - final Cast<SourceRecord> xform = new Cast.Value<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int8")); - xform.apply(new SourceRecord(null, null, "topic", 0, null, Collections.singletonList("foo"))); + xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int8")); + xformValue.apply(new SourceRecord(null, null, "topic", 0, null, Collections.singletonList("foo"))); } @Test public void castFieldsWithSchema() { - final Cast<SourceRecord> xform = new Cast.Value<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int8:int16,int16:int32,int32:int64,int64:boolean,float32:float64,float64:boolean,boolean:int8,string:int32,optional:int32")); + xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int8:int16,int16:int32,int32:int64,int64:boolean,float32:float64,float64:boolean,boolean:int8,string:int32,optional:int32")); // Include an optional fields and fields with defaults to validate their values are passed through properly SchemaBuilder builder = SchemaBuilder.struct(); @@ -336,7 +317,7 @@ public class CastTest { recordValue.put("string", "42"); // optional field intentionally omitted - SourceRecord transformed = xform.apply(new SourceRecord(null, null, "topic", 0, + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, supportedTypesSchema, recordValue)); assertEquals((short) 8, ((Struct) transformed.value()).get("int8")); @@ -356,8 +337,7 @@ public class CastTest { @SuppressWarnings("unchecked") @Test public void castFieldsSchemaless() { - final Cast<SourceRecord> xform = new Cast.Value<>(); - xform.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int8:int16,int16:int32,int32:int64,int64:boolean,float32:float64,float64:boolean,boolean:int8,string:int32")); + xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int8:int16,int16:int32,int32:int64,int64:boolean,float32:float64,float64:boolean,boolean:int8,string:int32")); Map<String, Object> recordValue = new HashMap<>(); recordValue.put("int8", (byte) 8); recordValue.put("int16", (short) 16); @@ -367,7 +347,7 @@ public class CastTest { recordValue.put("float64", -64.); recordValue.put("boolean", true); recordValue.put("string", "42"); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, "topic", 0, + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, null, recordValue)); assertNull(transformed.valueSchema()); http://git-wip-us.apache.org/repos/asf/kafka/blob/f87d58b7/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/ExtractFieldTest.java ---------------------------------------------------------------------- diff --git a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/ExtractFieldTest.java b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/ExtractFieldTest.java index b54a908..0b7ce96 100644 --- a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/ExtractFieldTest.java +++ b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/ExtractFieldTest.java @@ -20,6 +20,7 @@ import org.apache.kafka.connect.data.Schema; import org.apache.kafka.connect.data.SchemaBuilder; import org.apache.kafka.connect.data.Struct; import org.apache.kafka.connect.sink.SinkRecord; +import org.junit.After; import org.junit.Test; import java.util.Collections; @@ -28,10 +29,15 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; public class ExtractFieldTest { + private final ExtractField<SinkRecord> xform = new ExtractField.Key<>(); + + @After + public void teardown() { + xform.close(); + } @Test public void schemaless() { - final ExtractField<SinkRecord> xform = new ExtractField.Key<>(); xform.configure(Collections.singletonMap("field", "magic")); final SinkRecord record = new SinkRecord("test", 0, null, Collections.singletonMap("magic", 42), null, null, 0); @@ -43,7 +49,6 @@ public class ExtractFieldTest { @Test public void withSchema() { - final ExtractField<SinkRecord> xform = new ExtractField.Key<>(); xform.configure(Collections.singletonMap("field", "magic")); final Schema keySchema = SchemaBuilder.struct().field("magic", Schema.INT32_SCHEMA).build(); http://git-wip-us.apache.org/repos/asf/kafka/blob/f87d58b7/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/FlattenTest.java ---------------------------------------------------------------------- diff --git a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/FlattenTest.java b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/FlattenTest.java index 86851f3..d709054 100644 --- a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/FlattenTest.java +++ b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/FlattenTest.java @@ -22,6 +22,7 @@ import org.apache.kafka.connect.data.SchemaBuilder; import org.apache.kafka.connect.data.Struct; import org.apache.kafka.connect.errors.DataException; import org.apache.kafka.connect.source.SourceRecord; +import org.junit.After; import org.junit.Test; import java.util.Arrays; @@ -36,25 +37,30 @@ import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; public class FlattenTest { + private final Flatten<SourceRecord> xformKey = new Flatten.Key<>(); + private final Flatten<SourceRecord> xformValue = new Flatten.Value<>(); + + @After + public void teardown() { + xformKey.close(); + xformValue.close(); + } @Test(expected = DataException.class) public void topLevelStructRequired() { - final Flatten<SourceRecord> xform = new Flatten.Value<>(); - xform.configure(Collections.<String, String>emptyMap()); - xform.apply(new SourceRecord(null, null, "topic", 0, Schema.INT32_SCHEMA, 42)); + xformValue.configure(Collections.<String, String>emptyMap()); + xformValue.apply(new SourceRecord(null, null, "topic", 0, Schema.INT32_SCHEMA, 42)); } @Test(expected = DataException.class) public void topLevelMapRequired() { - final Flatten<SourceRecord> xform = new Flatten.Value<>(); - xform.configure(Collections.<String, String>emptyMap()); - xform.apply(new SourceRecord(null, null, "topic", 0, null, 42)); + xformValue.configure(Collections.<String, String>emptyMap()); + xformValue.apply(new SourceRecord(null, null, "topic", 0, null, 42)); } @Test public void testNestedStruct() { - final Flatten<SourceRecord> xform = new Flatten.Value<>(); - xform.configure(Collections.<String, String>emptyMap()); + xformValue.configure(Collections.<String, String>emptyMap()); SchemaBuilder builder = SchemaBuilder.struct(); builder.field("int8", Schema.INT8_SCHEMA); @@ -93,7 +99,7 @@ public class FlattenTest { Struct twoLevelNestedStruct = new Struct(twoLevelNestedSchema); twoLevelNestedStruct.put("A", oneLevelNestedStruct); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, twoLevelNestedSchema, twoLevelNestedStruct)); @@ -113,8 +119,7 @@ public class FlattenTest { @Test public void testNestedMapWithDelimiter() { - final Flatten<SourceRecord> xform = new Flatten.Value<>(); - xform.configure(Collections.singletonMap("delimiter", "#")); + xformValue.configure(Collections.singletonMap("delimiter", "#")); Map<String, Object> supportedTypes = new HashMap<>(); supportedTypes.put("int8", (byte) 8); @@ -130,7 +135,7 @@ public class FlattenTest { Map<String, Object> oneLevelNestedMap = Collections.singletonMap("B", (Object) supportedTypes); Map<String, Object> twoLevelNestedMap = Collections.singletonMap("A", (Object) oneLevelNestedMap); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, null, twoLevelNestedMap)); @@ -151,8 +156,7 @@ public class FlattenTest { @Test public void testOptionalFieldStruct() { - final Flatten<SourceRecord> xform = new Flatten.Value<>(); - xform.configure(Collections.<String, String>emptyMap()); + xformValue.configure(Collections.<String, String>emptyMap()); SchemaBuilder builder = SchemaBuilder.struct(); builder.field("opt_int32", Schema.OPTIONAL_INT32_SCHEMA); @@ -168,7 +172,7 @@ public class FlattenTest { Struct oneLevelNestedStruct = new Struct(oneLevelNestedSchema); oneLevelNestedStruct.put("B", supportedTypes); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, oneLevelNestedSchema, oneLevelNestedStruct)); @@ -179,15 +183,14 @@ public class FlattenTest { @Test public void testOptionalFieldMap() { - final Flatten<SourceRecord> xform = new Flatten.Value<>(); - xform.configure(Collections.<String, String>emptyMap()); + xformValue.configure(Collections.<String, String>emptyMap()); Map<String, Object> supportedTypes = new HashMap<>(); supportedTypes.put("opt_int32", null); Map<String, Object> oneLevelNestedMap = Collections.singletonMap("B", (Object) supportedTypes); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, null, oneLevelNestedMap)); @@ -200,12 +203,11 @@ public class FlattenTest { @Test public void testKey() { - final Flatten<SourceRecord> xform = new Flatten.Key<>(); - xform.configure(Collections.<String, String>emptyMap()); + xformKey.configure(Collections.<String, String>emptyMap()); Map<String, Map<String, Integer>> key = Collections.singletonMap("A", Collections.singletonMap("B", 12)); SourceRecord src = new SourceRecord(null, null, "topic", null, key, null, null); - SourceRecord transformed = xform.apply(src); + SourceRecord transformed = xformKey.apply(src); assertNull(transformed.keySchema()); assertTrue(transformed.key() instanceof Map); @@ -215,10 +217,9 @@ public class FlattenTest { @Test(expected = DataException.class) public void testUnsupportedTypeInMap() { - final Flatten<SourceRecord> xform = new Flatten.Value<>(); - xform.configure(Collections.<String, String>emptyMap()); + xformValue.configure(Collections.<String, String>emptyMap()); Object value = Collections.singletonMap("foo", Arrays.asList("bar", "baz")); - xform.apply(new SourceRecord(null, null, "topic", 0, null, value)); + xformValue.apply(new SourceRecord(null, null, "topic", 0, null, value)); } @Test @@ -227,8 +228,7 @@ public class FlattenTest { // children should also be optional. Similarly, if the parent Struct has a default value, the default value for // the flattened field - final Flatten<SourceRecord> xform = new Flatten.Value<>(); - xform.configure(Collections.<String, String>emptyMap()); + xformValue.configure(Collections.<String, String>emptyMap()); SchemaBuilder builder = SchemaBuilder.struct().optional(); builder.field("req_field", Schema.STRING_SCHEMA); @@ -240,7 +240,7 @@ public class FlattenTest { // Intentionally leave this entire value empty since it is optional Struct value = new Struct(schema); - SourceRecord transformed = xform.apply(new SourceRecord(null, null, "topic", 0, schema, value)); + SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0, schema, value)); assertNotNull(transformed); Schema transformedSchema = transformed.valueSchema(); http://git-wip-us.apache.org/repos/asf/kafka/blob/f87d58b7/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/HoistFieldTest.java ---------------------------------------------------------------------- diff --git a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/HoistFieldTest.java b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/HoistFieldTest.java index 299aab3..1135b85 100644 --- a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/HoistFieldTest.java +++ b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/HoistFieldTest.java @@ -19,6 +19,7 @@ package org.apache.kafka.connect.transforms; import org.apache.kafka.connect.data.Schema; import org.apache.kafka.connect.data.Struct; import org.apache.kafka.connect.sink.SinkRecord; +import org.junit.After; import org.junit.Test; import java.util.Collections; @@ -27,10 +28,15 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; public class HoistFieldTest { + private final HoistField<SinkRecord> xform = new HoistField.Key<>(); + + @After + public void teardown() { + xform.close(); + } @Test public void schemaless() { - final HoistField<SinkRecord> xform = new HoistField.Key<>(); xform.configure(Collections.singletonMap("field", "magic")); final SinkRecord record = new SinkRecord("test", 0, null, 42, null, null, 0); @@ -42,7 +48,6 @@ public class HoistFieldTest { @Test public void withSchema() { - final HoistField<SinkRecord> xform = new HoistField.Key<>(); xform.configure(Collections.singletonMap("field", "magic")); final SinkRecord record = new SinkRecord("test", 0, Schema.INT32_SCHEMA, 42, null, null, 0); http://git-wip-us.apache.org/repos/asf/kafka/blob/f87d58b7/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/InsertFieldTest.java ---------------------------------------------------------------------- diff --git a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/InsertFieldTest.java b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/InsertFieldTest.java index 4ce6ad4..a0a0975 100644 --- a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/InsertFieldTest.java +++ b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/InsertFieldTest.java @@ -22,6 +22,7 @@ import org.apache.kafka.connect.data.Struct; import org.apache.kafka.connect.data.Timestamp; import org.apache.kafka.connect.errors.DataException; import org.apache.kafka.connect.source.SourceRecord; +import org.junit.After; import org.junit.Test; import java.util.Collections; @@ -32,14 +33,17 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertSame; public class InsertFieldTest { + private InsertField<SourceRecord> xform = new InsertField.Value<>(); + + @After + public void teardown() { + xform.close(); + } @Test(expected = DataException.class) public void topLevelStructRequired() { - final InsertField<SourceRecord> xform = new InsertField.Value<>(); xform.configure(Collections.singletonMap("topic.field", "topic_field")); - xform.apply(new SourceRecord(null, null, - "", 0, - Schema.INT32_SCHEMA, 42)); + xform.apply(new SourceRecord(null, null, "", 0, Schema.INT32_SCHEMA, 42)); } @Test @@ -51,7 +55,6 @@ public class InsertFieldTest { props.put("static.field", "instance_id"); props.put("static.value", "my-instance-id"); - final InsertField<SourceRecord> xform = new InsertField.Value<>(); xform.configure(props); final Schema simpleStructSchema = SchemaBuilder.struct().name("name").version(1).doc("doc").field("magic", Schema.OPTIONAL_INT64_SCHEMA).build(); @@ -94,7 +97,6 @@ public class InsertFieldTest { props.put("static.field", "instance_id"); props.put("static.value", "my-instance-id"); - final InsertField<SourceRecord> xform = new InsertField.Value<>(); xform.configure(props); final SourceRecord record = new SourceRecord(null, null, "test", 0, http://git-wip-us.apache.org/repos/asf/kafka/blob/f87d58b7/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/RegexRouterTest.java ---------------------------------------------------------------------- diff --git a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/RegexRouterTest.java b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/RegexRouterTest.java index cc001f1..aa47001 100644 --- a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/RegexRouterTest.java +++ b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/RegexRouterTest.java @@ -32,8 +32,9 @@ public class RegexRouterTest { props.put("replacement", replacement); final RegexRouter<SinkRecord> router = new RegexRouter<>(); router.configure(props); - return router.apply(new SinkRecord(topic, 0, null, null, null, null, 0)) - .topic(); + String sinkTopic = router.apply(new SinkRecord(topic, 0, null, null, null, null, 0)).topic(); + router.close(); + return sinkTopic; } @Test http://git-wip-us.apache.org/repos/asf/kafka/blob/f87d58b7/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/ReplaceFieldTest.java ---------------------------------------------------------------------- diff --git a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/ReplaceFieldTest.java b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/ReplaceFieldTest.java index e3d9d3a..6a1a13a 100644 --- a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/ReplaceFieldTest.java +++ b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/ReplaceFieldTest.java @@ -20,6 +20,7 @@ import org.apache.kafka.connect.data.Schema; import org.apache.kafka.connect.data.SchemaBuilder; import org.apache.kafka.connect.data.Struct; import org.apache.kafka.connect.sink.SinkRecord; +import org.junit.After; import org.junit.Test; import java.util.HashMap; @@ -28,11 +29,15 @@ import java.util.Map; import static org.junit.Assert.assertEquals; public class ReplaceFieldTest { + private ReplaceField<SinkRecord> xform = new ReplaceField.Value<>(); + + @After + public void teardown() { + xform.close(); + } @Test public void schemaless() { - final ReplaceField<SinkRecord> xform = new ReplaceField.Value<>(); - final Map<String, String> props = new HashMap<>(); props.put("blacklist", "dont"); props.put("renames", "abc:xyz,foo:bar"); @@ -57,8 +62,6 @@ public class ReplaceFieldTest { @Test public void withSchema() { - final ReplaceField<SinkRecord> xform = new ReplaceField.Value<>(); - final Map<String, String> props = new HashMap<>(); props.put("whitelist", "abc,foo"); props.put("renames", "abc:xyz,foo:bar"); http://git-wip-us.apache.org/repos/asf/kafka/blob/f87d58b7/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/SetSchemaMetadataTest.java ---------------------------------------------------------------------- diff --git a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/SetSchemaMetadataTest.java b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/SetSchemaMetadataTest.java index 206c51e..257b382 100644 --- a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/SetSchemaMetadataTest.java +++ b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/SetSchemaMetadataTest.java @@ -21,6 +21,7 @@ import org.apache.kafka.connect.data.Schema; import org.apache.kafka.connect.data.SchemaBuilder; import org.apache.kafka.connect.data.Struct; import org.apache.kafka.connect.sink.SinkRecord; +import org.junit.After; import org.junit.Test; import java.util.Collections; @@ -31,10 +32,15 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertSame; public class SetSchemaMetadataTest { + private final SetSchemaMetadata<SinkRecord> xform = new SetSchemaMetadata.Value<>(); + + @After + public void teardown() { + xform.close(); + } @Test public void schemaNameUpdate() { - final SetSchemaMetadata<SinkRecord> xform = new SetSchemaMetadata.Value<>(); xform.configure(Collections.singletonMap("schema.name", "foo")); final SinkRecord record = new SinkRecord("", 0, null, null, SchemaBuilder.struct().build(), null, 0); final SinkRecord updatedRecord = xform.apply(record); @@ -43,7 +49,6 @@ public class SetSchemaMetadataTest { @Test public void schemaVersionUpdate() { - final SetSchemaMetadata<SinkRecord> xform = new SetSchemaMetadata.Value<>(); xform.configure(Collections.singletonMap("schema.version", 42)); final SinkRecord record = new SinkRecord("", 0, null, null, SchemaBuilder.struct().build(), null, 0); final SinkRecord updatedRecord = xform.apply(record); @@ -56,7 +61,6 @@ public class SetSchemaMetadataTest { props.put("schema.name", "foo"); props.put("schema.version", "42"); - final SetSchemaMetadata<SinkRecord> xform = new SetSchemaMetadata.Value<>(); xform.configure(props); final SinkRecord record = new SinkRecord("", 0, null, null, SchemaBuilder.struct().build(), null, 0); @@ -83,7 +87,6 @@ public class SetSchemaMetadataTest { final Map<String, String> props = new HashMap<>(); props.put("schema.name", "foo"); props.put("schema.version", "42"); - final SetSchemaMetadata<SinkRecord> xform = new SetSchemaMetadata.Value<>(); xform.configure(props); final SinkRecord record = new SinkRecord("", 0, null, null, schema, value, 0);
