root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 3 Packages Projects Releases Wiki Activity

MINOR: clean up unneeded `@SuppressWarnings` (#10855) 

Reviewers: Luke Chen <showuon@gmail.com>, Matthias J. Sax <mjsax@apache.org>, Chia-Ping Tsai <chia7712@gmail.com>
This commit is contained in:
Josep Prat 2021-06-13 13:00:14 +02:00 committed by GitHub
parent 6de37e536a
commit 787b4fe955
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
33 changed files with 2 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
View File

@ -135,7 +135,6 @@ import static org.apache.kafka.common.config.ConfigDef.parseType;
* @see ConsumerConfig
* @see ProducerConfig
*/
@SuppressWarnings("deprecation")
public class StreamsConfig extends AbstractConfig {
private static final Logger log = LoggerFactory.getLogger(StreamsConfig.class);

1
streams/src/main/java/org/apache/kafka/streams/kstream/internals/CogroupedStreamAggregateBuilder.java
View File

@ -48,7 +48,6 @@ class CogroupedStreamAggregateBuilder<K, VOut> {
CogroupedStreamAggregateBuilder(final InternalStreamsBuilder builder) {
this.builder = builder;
}
@SuppressWarnings("unchecked")
<KR> KTable<KR, VOut> build(final Map<KGroupedStreamImpl<K, ?>, Aggregator<? super K, ? super Object, VOut>> groupPatterns,
final Initializer<VOut> initializer,
final NamedInternal named,

1
streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/KTableKTableJoinNode.java
View File

@ -209,7 +209,6 @@ public class KTableKTableJoinNode<K, V1, V2, VR> extends BaseJoinProcessorNode<K
return this;
}
@SuppressWarnings("unchecked")
public KTableKTableJoinNode<K, V1, V2, VR> build() {
return new KTableKTableJoinNode<>(
nodeName,

1
streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/ProcessorParameters.java
View File

@ -60,7 +60,6 @@ public class ProcessorParameters<KIn, VIn, KOut, VOut> {
return oldProcessorSupplier;
}
@SuppressWarnings("unchecked")
KTableSource<KIn, VIn> kTableSourceSupplier() {
// This cast always works because KTableSource hasn't been converted yet.
return oldProcessorSupplier == null

1
streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/TableProcessorNode.java
View File

@ -57,7 +57,6 @@ public class TableProcessorNode<K, V> extends GraphNode {
"} " + super.toString();
}
@SuppressWarnings("unchecked")
@Override
public void writeToTopology(final InternalTopologyBuilder topologyBuilder, final Properties props) {
final String processorName = processorParameters.processorName();

1
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorAdapter.java
View File

@ -46,7 +46,6 @@ public final class ProcessorAdapter<KIn, VIn, KOut, VOut> implements Processor<K
this.delegate = delegate;
}
@SuppressWarnings("unchecked")
@Override
public void init(final ProcessorContext<KOut, VOut> context) {
// It only makes sense to use this adapter internally to Streams, in which case

1
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java
View File

@ -263,7 +263,6 @@ public class ProcessorContextImpl extends AbstractProcessorContext<Object, Objec
streamTask.requestCommit();
}
@SuppressWarnings("deprecation") // removing #schedule(final long intervalMs,...) will fix this
@Override
public Cancellable schedule(final Duration interval,
final PunctuationType type,

1
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordDeserializer.java
View File

@ -50,7 +50,6 @@ class RecordDeserializer {
* {@link DeserializationExceptionHandler.DeserializationHandlerResponse#FAIL FAIL}
* or throws an exception itself
*/
@SuppressWarnings("deprecation")
ConsumerRecord<Object, Object> deserialize(final ProcessorContext processorContext,
final ConsumerRecord<byte[], byte[]> rawRecord) {

1
streams/src/main/java/org/apache/kafka/streams/state/internals/RecordConverters.java
View File

@ -23,7 +23,6 @@ import java.nio.ByteBuffer;
public final class RecordConverters {
private static final RecordConverter IDENTITY_INSTANCE = record -> record;
@SuppressWarnings("deprecation")
private static final RecordConverter RAW_TO_TIMESTAMED_INSTANCE = record -> {
final byte[] rawValue = record.value();
final long timestamp = record.timestamp();

3
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedWindowStore.java
View File

@ -66,7 +66,6 @@ public class RocksDBTimeOrderedWindowStore
throw new UnsupportedOperationException();
}
@SuppressWarnings("deprecation")
@Override
public WindowStoreIterator<byte[]> fetch(final Bytes key, final long timeFrom, final long timeTo) {
throw new UnsupportedOperationException();
@ -77,7 +76,6 @@ public class RocksDBTimeOrderedWindowStore
throw new UnsupportedOperationException();
}
@SuppressWarnings("deprecation") // note, this method must be kept if super#fetch(...) is removed
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes keyFrom,
final Bytes keyTo,
@ -105,7 +103,6 @@ public class RocksDBTimeOrderedWindowStore
throw new UnsupportedOperationException();
}
@SuppressWarnings("deprecation") // note, this method must be kept if super#fetchAll(...) is removed
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> fetchAll(final long timeFrom, final long timeTo) {
throw new UnsupportedOperationException();

3
streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedWindowStoreBuilder.java
View File

@ -135,7 +135,6 @@ public class TimestampedWindowStoreBuilder<K, V>
return wrapped.fetch(key, time);
}
@SuppressWarnings("deprecation")
@Override
public WindowStoreIterator<byte[]> fetch(final Bytes key,
final long timeFrom,
@ -150,7 +149,6 @@ public class TimestampedWindowStoreBuilder<K, V>
return wrapped.backwardFetch(key, timeFrom, timeTo);
}
@SuppressWarnings("deprecation")
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes keyFrom,
final Bytes keyTo,
@ -167,7 +165,6 @@ public class TimestampedWindowStoreBuilder<K, V>
return wrapped.backwardFetch(keyFrom, keyTo, timeFrom, timeTo);
}
@SuppressWarnings("deprecation")
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> fetchAll(final long timeFrom,
final long timeTo) {

3
streams/src/main/java/org/apache/kafka/streams/state/internals/WindowToTimestampedWindowByteStoreAdapter.java
View File

@ -54,7 +54,6 @@ class WindowToTimestampedWindowByteStoreAdapter implements WindowStore<Bytes, by
}
@Override
@SuppressWarnings("deprecation")
public WindowStoreIterator<byte[]> fetch(final Bytes key,
final long timeFrom,
final long timeTo) {
@ -83,7 +82,6 @@ class WindowToTimestampedWindowByteStoreAdapter implements WindowStore<Bytes, by
}
@Override
@SuppressWarnings("deprecation")
public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes keyFrom,
final Bytes keyTo,
final long timeFrom,
@ -126,7 +124,6 @@ class WindowToTimestampedWindowByteStoreAdapter implements WindowStore<Bytes, by
}
@Override
@SuppressWarnings("deprecation")
public KeyValueIterator<Windowed<Bytes>, byte[]> fetchAll(final long timeFrom,
final long timeTo) {
return new KeyValueToTimestampedKeyValueIteratorAdapter<>(store.fetchAll(timeFrom, timeTo));

1
streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
View File

@ -1079,7 +1079,6 @@ public class KafkaStreamsTest {
}
}
@SuppressWarnings("unchecked")
@Deprecated // testing old PAPI
private Topology getStatefulTopology(final String inputTopic,
final String outputTopic,

1
streams/src/test/java/org/apache/kafka/streams/integration/EosV2UpgradeIntegrationTest.java
View File

@ -875,7 +875,6 @@ public class EosV2UpgradeIntegrationTest {
final KStream<Long, Long> input = builder.stream(MULTI_PARTITION_INPUT_TOPIC);
input.transform(new TransformerSupplier<Long, Long, KeyValue<Long, Long>>() {
@SuppressWarnings("unchecked")
@Override
public Transformer<Long, Long, KeyValue<Long, Long>> get() {
return new Transformer<Long, Long, KeyValue<Long, Long>>() {

1
streams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskCreationIntegrationTest.java
View File

@ -105,7 +105,6 @@ public class StandbyTaskCreationIntegrationTest {
builder.addStateStore(keyValueStoreBuilder);
builder.stream(INPUT_TOPIC, Consumed.with(Serdes.Integer(), Serdes.Integer()))
.transform(() -> new Transformer<Integer, Integer, KeyValue<Integer, Integer>>() {
@SuppressWarnings("unchecked")
@Override
public void init(final ProcessorContext context) {}

1
streams/src/test/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilderTest.java
View File

@ -51,7 +51,6 @@ import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
@SuppressWarnings("unchecked")
public class InternalStreamsBuilderTest {
private static final String APP_ID = "app-id";

3
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KGroupedTableImplTest.java
View File

@ -201,7 +201,6 @@ public class KGroupedTableImplTest {
}
}
@SuppressWarnings("unchecked")
@Test
public void shouldReduceAndMaterializeResults() {
final KeyValueMapper<String, Number, KeyValue<String, Integer>> intProjection =
@ -235,7 +234,6 @@ public class KGroupedTableImplTest {
}
}
@SuppressWarnings("unchecked")
@Test
public void shouldCountAndMaterializeResults() {
builder
@ -265,7 +263,6 @@ public class KGroupedTableImplTest {
}
}
@SuppressWarnings("unchecked")
@Test
public void shouldAggregateAndMaterializeResults() {
builder

1
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableImplTest.java
View File

@ -574,7 +574,6 @@ public class KTableImplTest {
assertThrows(NullPointerException.class, () -> table.transformValues(valueTransformerSupplier, (Materialized) null));
}
@SuppressWarnings("unchecked")
@Test
public void shouldThrowNullPointerOnTransformValuesWithKeyWhenStoreNamesNull() {
final ValueTransformerWithKeySupplier<String, String, ?> valueTransformerSupplier =

1
streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionWindowedKStreamImplTest.java
View File

@ -283,7 +283,6 @@ public class SessionWindowedKStreamImplTest {
}
@Test
@SuppressWarnings("unchecked")
public void shouldThrowNullPointerOnMaterializedReduceIfNamedIsNull() {
assertThrows(NullPointerException.class, () -> stream.reduce(MockReducer.STRING_ADDER, (Named) null));
}

1
streams/src/test/java/org/apache/kafka/streams/kstream/internals/SlidingWindowedKStreamImplTest.java
View File

@ -361,7 +361,6 @@ public class SlidingWindowedKStreamImplTest {
}
@Test
@SuppressWarnings("unchecked")
public void shouldThrowNullPointerOnMaterializedReduceIfNamedIsNull() {
assertThrows(NullPointerException.class, () -> windowedStream.reduce(MockReducer.STRING_ADDER, (Named) null));
}

1
streams/src/test/java/org/apache/kafka/streams/kstream/internals/TimeWindowedKStreamImplTest.java
View File

@ -300,7 +300,6 @@ public class TimeWindowedKStreamImplTest {
}
@Test
@SuppressWarnings("unchecked")
public void shouldThrowNullPointerOnMaterializedReduceIfNamedIsNull() {
assertThrows(NullPointerException.class, () -> windowedStream.reduce(
MockReducer.STRING_ADDER,

1
streams/src/test/java/org/apache/kafka/streams/kstream/internals/TransformerSupplierAdapterTest.java
View File

@ -34,7 +34,6 @@ import static org.hamcrest.core.IsSame.sameInstance;
import static org.hamcrest.core.IsNot.not;
import static org.hamcrest.MatcherAssert.assertThat;
@SuppressWarnings("unchecked")
public class TransformerSupplierAdapterTest extends EasyMockSupport {
private ProcessorContext context;

1
streams/src/test/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionResponseWrapperSerdeTest.java
View File

@ -127,7 +127,6 @@ public class SubscriptionResponseWrapperSerdeTest {
}
@Test
@SuppressWarnings("unchecked")
public void shouldThrowExceptionWithBadVersionTest() {
final long[] hashedValue = null;
assertThrows(UnsupportedVersionException.class,

2
streams/src/test/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionWrapperSerdeTest.java
View File

@ -59,7 +59,6 @@ public class SubscriptionWrapperSerdeTest {
}
@Test
@SuppressWarnings("unchecked")
public void shouldThrowExceptionOnNullKeyTest() {
final String originalKey = null;
final long[] hashedValue = Murmur3.hash128(new byte[] {(byte) 0xFF, (byte) 0xAA, (byte) 0x00, (byte) 0x19});
@ -68,7 +67,6 @@ public class SubscriptionWrapperSerdeTest {
}
@Test
@SuppressWarnings("unchecked")
public void shouldThrowExceptionOnNullInstructionTest() {
final String originalKey = "originalKey";
final long[] hashedValue = Murmur3.hash128(new byte[] {(byte) 0xFF, (byte) 0xAA, (byte) 0x00, (byte) 0x19});

8
streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorContextImplTest.java
View File

@ -220,7 +220,6 @@ public class ProcessorContextImplTest {
}
@Test
@SuppressWarnings("deprecation")
public void globalWindowStoreShouldBeReadOnly() {
doTest("GlobalWindowStore", (Consumer<WindowStore<String, Long>>) store -> {
verifyStoreCannotBeInitializedOrClosed(store);
@ -238,7 +237,6 @@ public class ProcessorContextImplTest {
@Test
@SuppressWarnings("deprecation")
public void globalTimestampedWindowStoreShouldBeReadOnly() {
doTest("GlobalTimestampedWindowStore", (Consumer<TimestampedWindowStore<String, Long>>) store -> {
verifyStoreCannotBeInitializedOrClosed(store);
@ -325,7 +323,6 @@ public class ProcessorContextImplTest {
}
@Test
@SuppressWarnings("deprecation")
public void localWindowStoreShouldNotAllowInitOrClose() {
doTest("LocalWindowStore", (Consumer<WindowStore<String, Long>>) store -> {
verifyStoreCannotBeInitializedOrClosed(store);
@ -345,7 +342,6 @@ public class ProcessorContextImplTest {
}
@Test
@SuppressWarnings("deprecation")
public void localTimestampedWindowStoreShouldNotAllowInitOrClose() {
doTest("LocalTimestampedWindowStore", (Consumer<TimestampedWindowStore<String, Long>>) store -> {
verifyStoreCannotBeInitializedOrClosed(store);
@ -615,7 +611,7 @@ public class ProcessorContextImplTest {
return timestampedKeyValueStoreMock;
}
@SuppressWarnings({"unchecked", "deprecation"})
@SuppressWarnings("unchecked")
private WindowStore<String, Long> windowStoreMock() {
final WindowStore<String, Long> windowStore = mock(WindowStore.class);
@ -638,7 +634,7 @@ public class ProcessorContextImplTest {
return windowStore;
}
@SuppressWarnings({"unchecked", "deprecation"})
@SuppressWarnings("unchecked")
private TimestampedWindowStore<String, Long> timestampedWindowStoreMock() {
final TimestampedWindowStore<String, Long> windowStore = mock(TimestampedWindowStore.class);

12
streams/src/test/java/org/apache/kafka/streams/state/internals/CachingPersistentWindowStoreTest.java
View File

@ -236,7 +236,6 @@ public class CachingPersistentWindowStoreTest {
}
@Test
@SuppressWarnings("deprecation")
public void shouldPutFetchFromCache() {
cachingStore.put(bytesKey("a"), bytesValue("a"), DEFAULT_TIMESTAMP);
cachingStore.put(bytesKey("b"), bytesValue("b"), DEFAULT_TIMESTAMP);
@ -275,7 +274,6 @@ public class CachingPersistentWindowStoreTest {
}
@Test
@SuppressWarnings("deprecation")
public void shouldPutFetchRangeFromCache() {
cachingStore.put(bytesKey("a"), bytesValue("a"), DEFAULT_TIMESTAMP);
cachingStore.put(bytesKey("b"), bytesValue("b"), DEFAULT_TIMESTAMP);
@ -317,7 +315,6 @@ public class CachingPersistentWindowStoreTest {
}
@Test
@SuppressWarnings("deprecation")
public void shouldGetAllBackwardFromCache() {
cachingStore.put(bytesKey("a"), bytesValue("a"), DEFAULT_TIMESTAMP);
cachingStore.put(bytesKey("b"), bytesValue("b"), DEFAULT_TIMESTAMP);
@ -340,7 +337,6 @@ public class CachingPersistentWindowStoreTest {
}
@Test
@SuppressWarnings("deprecation")
public void shouldFetchAllWithinTimestampRange() {
final String[] array = {"a", "b", "c", "d", "e", "f", "g", "h"};
for (int i = 0; i < array.length; i++) {
@ -382,7 +378,6 @@ public class CachingPersistentWindowStoreTest {
}
@Test
@SuppressWarnings("deprecation")
public void shouldFetchAllBackwardWithinTimestampRange() {
final String[] array = {"a", "b", "c", "d", "e", "f", "g", "h"};
for (int i = 0; i < array.length; i++) {
@ -439,7 +434,6 @@ public class CachingPersistentWindowStoreTest {
}
@Test
@SuppressWarnings("deprecation")
public void shouldForwardDirtyItemsWhenFlushCalled() {
final Windowed<String> windowedKey =
new Windowed<>("1", new TimeWindow(DEFAULT_TIMESTAMP, DEFAULT_TIMESTAMP + WINDOW_SIZE));
@ -456,7 +450,6 @@ public class CachingPersistentWindowStoreTest {
}
@Test
@SuppressWarnings("deprecation")
public void shouldForwardOldValuesWhenEnabled() {
cachingStore.setFlushListener(cacheListener, true);
final Windowed<String> windowedKey =
@ -485,7 +478,6 @@ public class CachingPersistentWindowStoreTest {
}
@Test
@SuppressWarnings("deprecation")
public void shouldForwardOldValuesWhenDisabled() {
final Windowed<String> windowedKey =
new Windowed<>("1", new TimeWindow(DEFAULT_TIMESTAMP, DEFAULT_TIMESTAMP + WINDOW_SIZE));
@ -616,7 +608,6 @@ public class CachingPersistentWindowStoreTest {
}
@Test
@SuppressWarnings("deprecation")
public void shouldClearNamespaceCacheOnClose() {
cachingStore.put(bytesKey("a"), bytesValue("a"), 0L);
assertEquals(1, cache.size());
@ -637,7 +628,6 @@ public class CachingPersistentWindowStoreTest {
}
@Test
@SuppressWarnings("deprecation")
public void shouldThrowIfTryingToWriteToClosedCachingStore() {
cachingStore.close();
assertThrows(InvalidStateStoreException.class, () -> cachingStore.put(bytesKey("a"), bytesValue("a"), 0L));
@ -786,13 +776,11 @@ public class CachingPersistentWindowStoreTest {
}
@Test
@SuppressWarnings("deprecation")
public void shouldThrowNullPointerExceptionOnPutNullKey() {
assertThrows(NullPointerException.class, () -> cachingStore.put(null, bytesValue("anyValue"), 0L));
}
@Test
@SuppressWarnings("deprecation")
public void shouldNotThrowNullPointerExceptionOnPutNullValue() {
cachingStore.put(bytesKey("a"), null, 0L);
}

1
streams/src/test/java/org/apache/kafka/streams/state/internals/FilteredCacheIteratorTest.java
View File

@ -50,7 +50,6 @@ public class FilteredCacheIteratorTest {
}
};
@SuppressWarnings("unchecked")
private final KeyValueStore<Bytes, LRUCacheEntry> store = new GenericInMemoryKeyValueStore<>("my-store");
private final KeyValue<Bytes, LRUCacheEntry> firstEntry = KeyValue.pair(Bytes.wrap("a".getBytes()),
new LRUCacheEntry("1".getBytes()));

1
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreTest.java
View File

@ -426,7 +426,6 @@ public class MeteredTimestampedKeyValueStoreTest {
}
@Test
@SuppressWarnings("unchecked")
public void shouldNotThrowExceptionIfSerdesCorrectlySetFromProcessorContext() {
final MeteredTimestampedKeyValueStore<String, Long> store = new MeteredTimestampedKeyValueStore<>(
inner,

1
streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreTest.java
View File

@ -43,7 +43,6 @@ import static org.apache.kafka.test.StreamsTestUtils.valuesToSet;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@SuppressWarnings("PointlessArithmeticExpression")
public class RocksDBWindowStoreTest extends AbstractWindowBytesStoreTest {
private static final String STORE_NAME = "rocksDB window store";

1
streams/src/test/java/org/apache/kafka/streams/tests/StreamsUpgradeTest.java
View File

@ -65,7 +65,6 @@ import static org.apache.kafka.streams.processor.internals.assignment.StreamsAss
public class StreamsUpgradeTest {
@SuppressWarnings("unchecked")
public static void main(final String[] args) throws Exception {
if (args.length < 1) {
System.err.println("StreamsUpgradeTest requires one argument (properties-file) but no provided: ");

1
streams/src/test/java/org/apache/kafka/streams/tests/StreamsUpgradeToCooperativeRebalanceTest.java
View File

@ -37,7 +37,6 @@ import java.util.Set;
public class StreamsUpgradeToCooperativeRebalanceTest {
@SuppressWarnings("unchecked")
public static void main(final String[] args) throws Exception {
if (args.length < 1) {
System.err.println("StreamsUpgradeToCooperativeRebalanceTest requires one argument (properties-file) but no args provided");

1
streams/src/test/java/org/apache/kafka/test/GenericInMemoryKeyValueStore.java
View File

@ -60,7 +60,6 @@ public class GenericInMemoryKeyValueStore<K extends Comparable, V>
@Deprecated
@Override
@SuppressWarnings("unchecked")
/* This is a "dummy" store used for testing;
it does not support restoring from changelog since we allow it to be serde-ignorant */
public void init(final ProcessorContext context, final StateStore root) {

1
streams/src/test/java/org/apache/kafka/test/GenericInMemoryTimestampedKeyValueStore.java
View File

@ -61,7 +61,6 @@ public class GenericInMemoryTimestampedKeyValueStore<K extends Comparable, V>
@Deprecated
@Override
@SuppressWarnings("unchecked")
/* This is a "dummy" store used for testing;
it does not support restoring from changelog since we allow it to be serde-ignorant */
public void init(final ProcessorContext context, final StateStore root) {