Added timeouts on send and flush calls in KafkaProducerWrapper #696

jzakaryan · 2020-03-24T00:08:59Z

The changes in this PR address the issue of kafka client being stuck on send and flush in cases when the destination topic gets dropped. Since kafka treats missing topic metadata as an eventual consistency issue and keeps retrying to send hoping that the topic will be available again, we have to make our calls to those methods bounded (i.e. have them time out after a certain amount of time).
I use CompletableFutures to have timeouts on those calls which are non blocking. Seems like that's the best way to do that in java without having to resort to third party libraries.

…nfigurable timeouts

somandal

This is only a partial review, let's address these and I'll look deeper into the other parts of the newer code.

somandal · 2020-04-01T15:45:56Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaProducerWrapper.java

+ /**
+  * There are two known cases that lead to IllegalStateException and we should retry:
+  *   (1) number of brokers is less than minISR
+  *   (2) producer is closed in generateSendFailure by another thread
+  *   (3) For either condition, we should retry as broker comes back healthy or producer is recreated
+  */


Any reason why you moved these comments out? Also, I see that you've made the last line into (3), whereas it just talks about the above two points?

+1

To Sonam's first point: I think those comments are closely related to the logic within the method since they discuss impl details like exception types.

somandal · 2020-04-01T15:49:11Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaProducerWrapper.java

-        Thread.sleep(_sendFailureRetryWaitTimeMs);
-      } catch (TimeoutException e) {
-        _log.warn("Kafka producer buffer is full, retry in {} ms.", _sendFailureRetryWaitTimeMs, e);
+      } catch (TimeoutException ex) {


Was there some reason why you had to reorder the catches here? In the previous code, we catch IllegalStateException, and then catch TimeoutException, and then catch KafkaException. Let's not reorder these unless there is a good reason to. It also becomes harder to review since I can't easily see the actual changes vs. reordering.

Also you renamed all exception 'e' to 'ex', is that necessary?

somandal · 2020-04-01T15:51:07Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaProducerWrapper.java

+      } catch (IllegalStateException ex) {
+        // The following exception should be quite rare as most exceptions will be throw async callback
+        _log.warn("Either send is called on a closed producer or broker count is less than minISR, retry in {} ms.",
+            _sendFailureRetryWaitTimeMs, ex);


Looks like the earlier IllegalStateException block would sleep and you've removed it. why?
Thread.sleep(_sendFailureRetryWaitTimeMs);

Good catch. This is definitely a bug I introduced. Will revert the entire exception handling piece in send(). Better leave it as it is.

somandal · 2020-04-01T15:55:42Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaProducerWrapper.java

+        Throwable rootCause = ExceptionUtils.getRootCause(ex);
+        if (numberOfAttempts > MAX_SEND_ATTEMPTS ||
+            (rootCause instanceof Error || rootCause instanceof RuntimeException)) {
+          // Set a max_send_attempts for KafkaException as it may be non-recoverable


This comment doesn't make sense here, move it back to outside the if condition?

somandal · 2020-04-01T15:56:23Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaProducerWrapper.java

-            onComplete.onCompletion(metadata, generateSendFailure(exception));
-          }
-        }));
+        numberOfAttempts++;


change this back to ++numberOfAttempts?

+1

I like the rename (added s) but prefer the prefix form.

In fact, it wouldn't be such a bad idea to turn that loop into a for loop:

for (int numberOfAttempts = 1; retry; ++numberOfAttempts) { ... }

or even eliminate retry entirely:

for (int numberOfAttempts = 1;; ++numberOfAttempts) { try { maybeGetKafkaProducer(task).ifPresent(p -> doSend(p, producerRecord, onComplete)); return; } catch (...) { ... } }

somandal · 2020-04-01T15:58:20Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaProducerWrapper.java

      }
    }
  }

-  private synchronized void shutdownProducer() {
+  void doSend(Producer<K, V> producer, ProducerRecord<K, V> record, Callback callback) {


is this intended to be package private? If so, add a comment, otherwise add the appropriate public/protected/private

@somandal
I think that's okay; this method's intended to be overriden by another class in the same package. If it's marked protected, it'll be more visible than it needs to be, private wouldn't allow it to be overriden, and public is too permissive.

Sure, makes sense. Just keep things consistent, i.e. if you need a more visible scope for multiple methods, declare them all as package private, and the rest as private. Don't mix protected and package private without having a very good reason to.

somandal · 2020-04-01T15:59:15Z

config/server.properties

+brooklin.server.kafkaProducerWrapper.sendTimeout=50000
+brooklin.server.kafkaProducerWrapper.flushTimeout=3600000


This wouldn't be the right way to pass configs to your KafkaProducerWrapper. Lets discuss how to do this offline

somandal · 2020-04-01T16:01:17Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/BoundedKafkaProducerWrapper.java

+  private static final String SEND_TIMEOUT_CONFIG_KEY = "brooklin.server.kafkaProducerWrapper.sendTimeout";
+  private static final String FLUSH_TIMEOUT_CONFIG_KEY = "brooklin.server.kafkaProducerWrapper.flushTimeout";


This isn't the right way to pass configs. Any configs you want to pass here need to be scoped under the transport provider configs. The KafkaProducerWrapper receives all of the transport provider configs, so you'd want to add these under that scope. We don't need to access the full property name, because as the configs pass through the layers, the relevant prefixes are removed. You can see how other configs are accessed in KafkaProducerWrapper, you'll see they don't have that whole "brooklin.server" prefix. You can access these the same way.

@somandal I've discussed this with Ahmed. Will push the fixe soon

somandal · 2020-04-01T16:10:33Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaProducerWrapper.java

@@ -142,7 +143,7 @@ private void populateDefaultProducerConfigs() {
        DEFAULT_MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_VALUE);
  }

-  private Optional<Producer<K, V>> maybeGetKafkaProducer(DatastreamTask task) {
+  protected Optional<Producer<K, V>> maybeGetKafkaProducer(DatastreamTask task) {


I don't see you overriding this in your Bounded implementation, why make this protected?

somandal · 2020-04-01T16:10:43Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaProducerWrapper.java

@@ -246,7 +253,7 @@ private synchronized void shutdownProducer() {
    }
  }

-  private DatastreamRuntimeException generateSendFailure(Exception exception) {
+  protected DatastreamRuntimeException generateSendFailure(Exception exception) {


I don't see you overriding this in your Bounded implementation, why make this protected?

somandal · 2020-04-01T16:15:17Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaProducerWrapper.java

@@ -60,7 +61,8 @@

  private static final int TIME_OUT = 2000;
  private static final int MAX_SEND_ATTEMPTS = 10;
-  private final Logger _log;
+
+  protected final Logger _log;


Move this with other protected member variables

ahmedahamid · 2020-04-01T21:01:16Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaProducerWrapper.java

@@ -72,8 +74,7 @@
  // Producer is lazily initialized during the first send call.
  // Also, can be nullified in case of exceptions, and recreated by subsequent send calls.
  // Mark as volatile as it is mutable and used by different threads
-  private volatile Producer<K, V> _kafkaProducer;
-
+  protected volatile Producer<K, V> _kafkaProducer;


It's not necessary to mark member fields/methods protected if the extenders of this class live in the same package. This is only useful if you want to make them accessible to extenders in different packages.

Since KafkaProducerWrapper and BoundedKafkaProducerWrapper both live in the same package (com.linkedin.datastream.kafka), all package-private (no modifier) fields/methods in the former are accessible/overridable to/by the latter.

ahmedahamid · 2020-04-01T21:05:54Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaProducerWrapper.java

+ /**
+  * There are two known cases that lead to IllegalStateException and we should retry:
+  *   (1) number of brokers is less than minISR
+  *   (2) producer is closed in generateSendFailure by another thread
+  *   (3) For either condition, we should retry as broker comes back healthy or producer is recreated
+  */


+1

To Sonam's first point: I think those comments are closely related to the logic within the method since they discuss impl details like exception types.

ahmedahamid · 2020-04-01T21:09:56Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaProducerWrapper.java

-            onComplete.onCompletion(metadata, generateSendFailure(exception));
-          }
-        }));
+        numberOfAttempts++;


+1

I like the rename (added s) but prefer the prefix form.

In fact, it wouldn't be such a bad idea to turn that loop into a for loop:

for (int numberOfAttempts = 1; retry; ++numberOfAttempts) { ... }

or even eliminate retry entirely:

for (int numberOfAttempts = 1;; ++numberOfAttempts) { try { maybeGetKafkaProducer(task).ifPresent(p -> doSend(p, producerRecord, onComplete)); return; } catch (...) { ... } }

ahmedahamid · 2020-04-01T21:39:42Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaProducerWrapper.java

@@ -17,6 +17,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Supplier;

+import org.apache.commons.lang.exception.ExceptionUtils;


Nothing wrong with this but I'd recommend lang3; it generally offers more modern facilities (better support for more recent versions of Java) and it's the one we have an explicit dependency on.

This would also entail adding an explicit dependency for this module on it in build.gradle :

project(':datastream-kafka') { dependencies { ... compile "org.apache.commons:commons-lang3:$commonslang3Version"

ahmedahamid · 2020-04-01T23:10:09Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaProducerWrapper.java

      }
    }
  }

-  private synchronized void shutdownProducer() {
+  void doSend(Producer<K, V> producer, ProducerRecord<K, V> record, Callback callback) {


@somandal
I think that's okay; this method's intended to be overriden by another class in the same package. If it's marked protected, it'll be more visible than it needs to be, private wouldn't allow it to be overriden, and public is too permissive.

ahmedahamid · 2020-04-02T03:56:07Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/BoundedKafkaProducerWrapper.java

+  private CompletableFuture<RecordMetadata> produceMessage(Producer<K, V> producer, ProducerRecord<K, V> record) {
+    CompletableFuture<RecordMetadata> future = new CompletableFuture<>();
+
+    producer.send(record, (metadata, exception) -> {
+      if (exception == null) {
+        future.complete(metadata);
+      } else {
+        future.completeExceptionally(new KafkaClientException(metadata, exception));
+      }
+    });
+
+    return future;
+  }


Thinking out loud:

This method is doing almost everything we need: it creates a CompletableFuture that is completed if the callback is called (send success/failure).

The only missing bit is canceling future after timeout elapses if future.isDone() is false, which can be accomplished with a ScheduledExecutorService. I know this is exactly what CompletableFutureUtils.failAfter() is doing but I think the logic over there is more than what's absolutely necessary; we don't really need the other CompletableFuture failAfter() creates or the additional logic in within(). We can just cancel this same future if it isn't done when timeout elapses.
scheduler.schedule(() -> { if (!future.isDone()) { future.cancel(); } }, _sendTimeout, TimeUnit.MILLISECONDS);

future.cancel() causes a java.util.concurrent.CancellationException to be thrown, which means we don't have to construct a TimeoutException ourselves because a CancellationException can only mean we cancelled it after the timeout elapsed.

This seems like something this method can do with a private executor service. I am not sure we really need a utils class just for this purpose.

@ahmedahamid I just think that cancellation and timeout are semantically different. We may want to cancel the future after timeout in our case, but that's not necessarily true in general. Also, smth may be cancelled without waiting for timeout (based on user input or other external factors). Just thinking out loud. Will see whether I can get rid of the utils.

I just think that cancellation and timeout are semantically different.

I didn't mean to suggest we should propagate CancellationException to callback. I was assuming doSend() will still construct a TimeoutException in case future is cancelled. That does sound a bit roundabout though; it would certainly be better to do future.completeExceptionally(new TimeoutException(...)) instead of future.cancel() if future.isDone() is false after timeout.

ahmedahamid · 2020-04-02T04:18:03Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/BoundedKafkaProducerWrapper.java

+      if (exception == null) {
+        future.complete(metadata);
+      } else {
+        future.completeExceptionally(new KafkaClientException(metadata, exception));


I don't think we need to propagate metadata if exception is null. The javadocs on Callback.onCompletion() state that metadata is null if an error occurred. This would also spare us having to introduce KafkaClientException.

There were tons of checks up the callback chain that are dealing with the metadata and exception. You're probably right, I need to see if it's safe to do so and remove KafkaClientException. I was kind of forced to introduce it in the first place.

ahmedahamid · 2020-04-02T06:09:58Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/BoundedKafkaProducerWrapper.java

+  synchronized void flush() {
+    if (_kafkaProducer != null) {
+      try {
+        CompletableFutureUtils.within(CompletableFuture.runAsync(() -> _kafkaProducer.flush()),


CompletableFuture.runAsync() uses the common pool which isn't the best option for a potentially long blocking call like Producer.flush(). Even if we provide our own thread pool to runAsync(), the CompletableFuture we'll get won't give us a way to interrupt a Producer.flush() call that exceeds the allowed timeout, which is necessary to free up the thread-pool thread in question. This is because calling cancel(true) on a CompletableFuture returned by runAsync() only causes a cancellation exception to be propagated without interrupting the blocked thread pool.

I'm afraid our only option here seems to be using an ExecutorService directly

// It's okay to use a single thread executor since flush() is synchronized ExecutorService executor = Executors.newSingleThreadExecutor(); Future<?> future = executor.submit(() -> super.flush()); try { // Block until timeout elapses future.get(_flushTimeout, TimeUnit.MILLISECONDS); } catch (TimeoutException e) { ... // Interrupt the Producer.flush() call to free up the blocked thread future.cancel(true); ... }

ahmedahamid · 2020-04-02T09:14:01Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/BoundedKafkaProducerWrapper.java

+  private static final int DEFAULT_SEND_TIME_OUT = 5000;
+  private static final int DEFAULT_FLUSH_TIME_OUT = 10 * 60 * 1000;


Please encode time units into variable/config names, e.g. DEFAULT_SEND_TIME_OUT_MS

Use longs for timeout configs (e.g. see existing timeout configs)

If you like, you can use Duration methods to initialize (e.g. Duration.ofSeconds(5).toMillis())

ahmedahamid · 2020-04-02T09:19:21Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/BoundedKafkaProducerWrapper.java

+  private int _sendTimeout;
+  private int _flushTimeout;


final

Add time unit suffixes (e.g. _sendTimoutMs)

ahmedahamid · 2020-04-02T09:49:11Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/BoundedKafkaProducerWrapper.java

+/**
+ * An extension of {@link KafkaProducerWrapper} with bounded calls for flush and send
+ */
+class BoundedKafkaProducerWrapper<K, V> extends KafkaProducerWrapper<K, V> {


Bounded is a little vague cause it's easy to confuse with buffering. I realize a better name won't be easy. If you can't think of one, just make sure the Javadoc is unambiguous (e.g. with timeouts for flush and send).

…erWrapper

vmaheshw · 2022-02-01T21:18:40Z

@jzakaryan Do we still need this PR?

Added timeouts on send and flush calls in KafkaProducerWrapper

fe19a50

jzakaryan requested review from ahmedahamid, celiakung and somandal March 24, 2020 00:08

Jhora Zakaryan added 3 commits March 24, 2020 14:09

Merge branch 'master' into FTR-KafkaClientTimeouts

abf10d5

Fixed conflicts and error handling logic in flush after merge

3134d6f

Made a separate abstraction for BoundedKafkaProducerWrapper. Added co…

988d8a5

…nfigurable timeouts

jzakaryan changed the title ~~[WIP] Added timeouts on send and flush calls in KafkaProducerWrapper~~ Added timeouts on send and flush calls in KafkaProducerWrapper Mar 30, 2020

somandal reviewed Apr 1, 2020

View reviewed changes

ahmedahamid reviewed Apr 2, 2020

View reviewed changes

Jhora Zakaryan added 2 commits April 7, 2020 15:06

Fixed access modifier issues. Handled flush cancellation better

1b8571e

Removed KafkaClientException class. Minor fixes in BoundedKafkaProduc…

9a11735

…erWrapper

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Added timeouts on send and flush calls in KafkaProducerWrapper #696

Added timeouts on send and flush calls in KafkaProducerWrapper #696

jzakaryan commented Mar 24, 2020 •

edited by ahmedahamid

somandal left a comment

somandal Apr 1, 2020

ahmedahamid Apr 1, 2020

somandal Apr 1, 2020

somandal Apr 1, 2020

jzakaryan Apr 7, 2020

somandal Apr 1, 2020

somandal Apr 1, 2020

ahmedahamid Apr 1, 2020

somandal Apr 1, 2020

ahmedahamid Apr 1, 2020

somandal Apr 2, 2020

somandal Apr 1, 2020

somandal Apr 1, 2020

jzakaryan Apr 1, 2020

somandal Apr 1, 2020

somandal Apr 1, 2020

somandal Apr 1, 2020

ahmedahamid Apr 1, 2020

ahmedahamid Apr 1, 2020

ahmedahamid Apr 1, 2020

ahmedahamid Apr 1, 2020

ahmedahamid Apr 1, 2020

ahmedahamid Apr 2, 2020 •

edited

jzakaryan Apr 7, 2020

ahmedahamid Apr 8, 2020

ahmedahamid Apr 2, 2020

jzakaryan Apr 7, 2020

ahmedahamid Apr 2, 2020 •

edited

ahmedahamid Apr 2, 2020 •

edited

ahmedahamid Apr 2, 2020 •

edited

ahmedahamid Apr 2, 2020 •

edited

vmaheshw commented Feb 1, 2022

		brooklin.server.kafkaProducerWrapper.sendTimeout=50000
		brooklin.server.kafkaProducerWrapper.flushTimeout=3600000

		private static final String SEND_TIMEOUT_CONFIG_KEY = "brooklin.server.kafkaProducerWrapper.sendTimeout";
		private static final String FLUSH_TIMEOUT_CONFIG_KEY = "brooklin.server.kafkaProducerWrapper.flushTimeout";

		private static final int DEFAULT_SEND_TIME_OUT = 5000;
		private static final int DEFAULT_FLUSH_TIME_OUT = 10 * 60 * 1000;

Added timeouts on send and flush calls in KafkaProducerWrapper #696

Are you sure you want to change the base?

Added timeouts on send and flush calls in KafkaProducerWrapper #696

Conversation

jzakaryan commented Mar 24, 2020 • edited by ahmedahamid

somandal left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ahmedahamid Apr 2, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ahmedahamid Apr 2, 2020 • edited

Choose a reason for hiding this comment

ahmedahamid Apr 2, 2020 • edited

Choose a reason for hiding this comment

ahmedahamid Apr 2, 2020 • edited

Choose a reason for hiding this comment

ahmedahamid Apr 2, 2020 • edited

Choose a reason for hiding this comment

vmaheshw commented Feb 1, 2022

jzakaryan commented Mar 24, 2020 •

edited by ahmedahamid

ahmedahamid Apr 2, 2020 •

edited

ahmedahamid Apr 2, 2020 •

edited

ahmedahamid Apr 2, 2020 •

edited

ahmedahamid Apr 2, 2020 •

edited

ahmedahamid Apr 2, 2020 •

edited