Skip to content

MINOR: improve Streams checkstyle and code cleanup #5954

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Merged
mjsax merged 6 commits into from
Dec 11, 2018
Merged

MINOR: improve Streams checkstyle and code cleanup #5954

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
258f42e
MINOR: improve Streams checkstyle and code cleanup
mjsax Nov 27, 2018
e826bc3
Github comments
mjsax Nov 28, 2018
966481e
Github comments
mjsax Nov 28, 2018
0040a8b
Rebased
mjsax Dec 6, 2018
9454584
Fix JavaDoc issues
mjsax Dec 8, 2018
c9c692f
Removed checkstyle rule
mjsax Dec 11, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions ...examples/src/main/java/org/apache/kafka/streams/examples/temperature/TemperatureDemo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@
*/
package org.apache.kafka.streams.examples.temperature;

import java.time.Duration;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
Expand All @@ -29,6 +28,7 @@
import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.kstream.WindowedSerdes;

import java.time.Duration;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;

Expand Down Expand Up @@ -90,10 +90,11 @@ public static void main(final String[] args) {
.groupByKey()
.windowedBy(TimeWindows.of(Duration.ofSeconds(TEMPERATURE_WINDOW_SIZE)))
.reduce((value1, value2) -> {
if (Integer.parseInt(value1) > Integer.parseInt(value2))
if (Integer.parseInt(value1) > Integer.parseInt(value2)) {
return value1;
else
} else {
return value2;
}
})
.toStream()
.filter((key, value) -> Integer.parseInt(value) > TEMPERATURE_THRESHOLD);
Expand Down
98 changes: 43 additions & 55 deletions streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@
*/
package org.apache.kafka.streams;

import java.time.Duration;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
Expand Down Expand Up @@ -63,6 +62,7 @@
import org.apache.kafka.streams.state.internals.StreamThreadStateStoreProvider;
import org.slf4j.Logger;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
Expand All @@ -77,7 +77,6 @@
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

import static org.apache.kafka.common.utils.Utils.getHost;
Expand Down Expand Up @@ -127,15 +126,13 @@
public class KafkaStreams implements AutoCloseable {

private static final String JMX_PREFIX = "kafka.streams";
private static final int DEFAULT_CLOSE_TIMEOUT = 0;

// processId is expected to be unique across JVMs and to be used
// in userData of the subscription request to allow assignor be aware
// of the co-location of stream thread's consumers. It is for internal
// usage only and should not be exposed to users at all.
private final Time time;
private final Logger log;
private final UUID processId;
private final String clientId;
private final Metrics metrics;
private final StreamsConfig config;
Expand Down Expand Up @@ -386,7 +383,9 @@ public void setGlobalStateRestoreListener(final StateRestoreListener globalState
result.putAll(thread.consumerMetrics());
result.putAll(thread.adminClientMetrics());
}
if (globalStreamThread != null) result.putAll(globalStreamThread.consumerMetrics());
if (globalStreamThread != null) {
result.putAll(globalStreamThread.consumerMetrics());
}
result.putAll(metrics.metrics());
return Collections.unmodifiableMap(result);
}
Expand Down Expand Up @@ -633,7 +632,7 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
this.time = time;

// The application ID is a required config and hence should always have value
processId = UUID.randomUUID();
final UUID processId = UUID.randomUUID();
final String userClientId = config.getString(StreamsConfig.CLIENT_ID_CONFIG);
final String applicationId = config.getString(StreamsConfig.APPLICATION_ID_CONFIG);
if (userClientId.length() <= 0) {
Expand Down Expand Up @@ -733,13 +732,10 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
final GlobalStateStoreProvider globalStateStoreProvider = new GlobalStateStoreProvider(internalTopologyBuilder.globalStateStores());
queryableStoreProvider = new QueryableStoreProvider(storeProviders, globalStateStoreProvider);

stateDirCleaner = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
@Override
public Thread newThread(final Runnable r) {
final Thread thread = new Thread(r, clientId + "-CleanupThread");
thread.setDaemon(true);
return thread;
}
stateDirCleaner = Executors.newSingleThreadScheduledExecutor(r -> {
final Thread thread = new Thread(r, clientId + "-CleanupThread");
thread.setDaemon(true);
return thread;
});
}

Expand Down Expand Up @@ -791,12 +787,9 @@ public synchronized void start() throws IllegalStateException, StreamsException
}

final Long cleanupDelay = config.getLong(StreamsConfig.STATE_CLEANUP_DELAY_MS_CONFIG);
stateDirCleaner.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
if (state == State.RUNNING) {
stateDirectory.cleanRemovedTasks(cleanupDelay);
}
stateDirCleaner.scheduleAtFixedRate(() -> {
if (state == State.RUNNING) {
stateDirectory.cleanRemovedTasks(cleanupDelay);
}
}, cleanupDelay, cleanupDelay, TimeUnit.MILLISECONDS);

Expand All @@ -814,7 +807,7 @@ public void run() {
* This will block until all threads have stopped.
*/
public void close() {
close(DEFAULT_CLOSE_TIMEOUT, TimeUnit.SECONDS);
close(Long.MAX_VALUE);

@mjsax mjsax Nov 28, 2018

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@ijuma Removed the SuppressWarning annotation and rewrote the code. Also have a PR for 2.1 branch: #5963 for this fix.

@mjsax mjsax Nov 28, 2018

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We translate old default of zero to Long.MAX_VALUE within deprecated close(final long timeout, final TimeUnit timeUnit) -- we can call private close() directly instead.

@ijuma ijuma Nov 29, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks, this looks better.

}

/**
Expand Down Expand Up @@ -856,45 +849,42 @@ private boolean close(final long timeoutMs) {
// wait for all threads to join in a separate thread;
// save the current thread so that if it is a stream thread
// we don't attempt to join it and cause a deadlock
final Thread shutdownThread = new Thread(new Runnable() {
@Override
public void run() {
// notify all the threads to stop; avoid deadlocks by stopping any
// further state reports from the thread since we're shutting down
for (final StreamThread thread : threads) {
thread.setStateListener(null);
thread.shutdown();
}
final Thread shutdownThread = new Thread(() -> {
// notify all the threads to stop; avoid deadlocks by stopping any
// further state reports from the thread since we're shutting down
for (final StreamThread thread : threads) {
thread.setStateListener(null);
thread.shutdown();
}

for (final StreamThread thread : threads) {
try {
if (!thread.isRunning()) {
thread.join();
}
} catch (final InterruptedException ex) {
Thread.currentThread().interrupt();
for (final StreamThread thread : threads) {
try {
if (!thread.isRunning()) {
thread.join();
}
} catch (final InterruptedException ex) {
Thread.currentThread().interrupt();
}
}

if (globalStreamThread != null) {
globalStreamThread.setStateListener(null);
globalStreamThread.shutdown();
}
if (globalStreamThread != null) {
globalStreamThread.setStateListener(null);
globalStreamThread.shutdown();
}

if (globalStreamThread != null && !globalStreamThread.stillRunning()) {
try {
globalStreamThread.join();
} catch (final InterruptedException e) {
Thread.currentThread().interrupt();
}
globalStreamThread = null;
if (globalStreamThread != null && !globalStreamThread.stillRunning()) {
try {
globalStreamThread.join();
} catch (final InterruptedException e) {
Thread.currentThread().interrupt();
}
globalStreamThread = null;
}

adminClient.close();
adminClient.close();

metrics.close();
setState(State.NOT_RUNNING);
}
metrics.close();
setState(State.NOT_RUNNING);
}, "kafka-streams-close-thread");

shutdownThread.setDaemon(true);
Expand All @@ -918,14 +908,12 @@ public void run() {
* @param timeout how long to wait for the threads to shutdown
* @return {@code true} if all threads were successfully stopped&mdash;{@code false} if the timeout was reached
* before all threads stopped
* Note that this method must not be called in the {@link StateListener#onChange(State, State)} callback of {@link StateListener}.
* Note that this method must not be called in the {@link StateListener#onChange(KafkaStreams.State, KafkaStreams.State)} callback of {@link StateListener}.
* @throws IllegalArgumentException if {@code timeout} can't be represented as {@code long milliseconds}
*/
public synchronized boolean close(final Duration timeout) throws IllegalArgumentException {
final String msgPrefix = prepareMillisCheckFailMsgPrefix(timeout, "timeout");
ApiUtils.validateMillisecondDuration(timeout, msgPrefix);

final long timeoutMs = timeout.toMillis();
final long timeoutMs = ApiUtils.validateMillisecondDuration(timeout, msgPrefix);
if (timeoutMs < 0) {
throw new IllegalArgumentException("Timeout can't be negative.");
}
Expand Down
3 changes: 2 additions & 1 deletion streams/src/main/java/org/apache/kafka/streams/KeyValue.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,8 +63,9 @@ public String toString() {

@Override
public boolean equals(final Object obj) {
if (this == obj)
if (this == obj) {
return true;
}

if (!(obj instanceof KeyValue)) {
return false;
Expand Down
6 changes: 4 additions & 2 deletions streams/src/main/java/org/apache/kafka/streams/internals/ApiUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,8 +36,9 @@ private ApiUtils() {
*/
public static long validateMillisecondDuration(final Duration duration, final String messagePrefix) {
try {
if (duration == null)
if (duration == null) {
throw new IllegalArgumentException(messagePrefix + "It shouldn't be null.");
}

return duration.toMillis();
} catch (final ArithmeticException e) {
Expand All @@ -53,8 +54,9 @@ public static long validateMillisecondDuration(final Duration duration, final St
*/
public static long validateMillisecondInstant(final Instant instant, final String messagePrefix) {
try {
if (instant == null)
if (instant == null) {
throw new IllegalArgumentException(messagePrefix + "It shouldn't be null.");
}

return instant.toEpochMilli();
} catch (final ArithmeticException e) {
Expand Down
4 changes: 2 additions & 2 deletions streams/src/main/java/org/apache/kafka/streams/kstream/Consumed.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -121,7 +121,7 @@ public static <K, V> Consumed<K, V> with(final TimestampExtractor timestampExtra
}

/**
* Create an instance of {@link Consumed} with a {@link Topology.AutoOffsetReset}.
* Create an instance of {@link Consumed} with a {@link org.apache.kafka.streams.Topology.AutoOffsetReset Topology.AutoOffsetReset}.
*
* @param resetPolicy the offset reset policy to be used. If {@code null} the default reset policy from config will be used
* @param <K> key type
Expand Down Expand Up @@ -166,7 +166,7 @@ public Consumed<K, V> withTimestampExtractor(final TimestampExtractor timestampE
}

/**
* Configure the instance of {@link Consumed} with a {@link Topology.AutoOffsetReset}.
* Configure the instance of {@link Consumed} with a {@link org.apache.kafka.streams.Topology.AutoOffsetReset Topology.AutoOffsetReset}.
*
* @param resetPolicy the offset reset policy to be used. If {@code null} the default reset policy from config will be used
* @return this
Expand Down
Loading