Mitigate errors in rebalances with partitions in SubSourceLogic #466

dvallejo · 2018-05-23T09:58:44Z

This fix mitigates the problem about close partitions with pending requests. It is related with this issue #382

To mitigate this problem I´ve wait a configured time before close the partition. This time can be configured with the property "wait-close-partition".
Also, I´ve added the possibility to re-use partitions. If we are going rebalance from 1 partition to 2, it´s not necessary close the original partition, just open a new one is necessary.

I had to ignore one IT because now a revoke is not required if we can re-use a partition.

I don´t have much experience with akka-streams or reactive kafka so any feedback is welcome.

lightbend-cla-validator · 2018-05-23T09:58:45Z

Hi @dvallejo,

Thank you for your contribution! We really value the time you've taken to put this together.

Before we proceed with reviewing this pull request, please sign the Lightbend Contributors License Agreement:

http://www.lightbend.com/contribute/cla

dvallejo · 2018-05-23T10:01:19Z

CLA signed

ennru

Thank you for trying to fix this!
Looks quite reasonable.

ennru · 2018-05-23T13:19:34Z

core/src/main/resources/reference.conf

@@ -84,6 +84,9 @@ akka.kafka.consumer {
    # Disable auto-commit by default
    enable.auto.commit = false
  }
+
+  # Time to wait for pending requests when a partition is closed
+  wait-close-partition = 5s


5 seconds as default sounds like a very long time.

Totally agree. Changed to 500 ms

ennru · 2018-05-23T13:19:42Z

core/src/main/scala/akka/kafka/internal/SubSourceLogic.scala

+    partitionsToRevoke = partitionsToRevoke -- tps
+
+    getOffsetsOnAssign.fold(pumpCB.invoke(topicsToBeAssigned)) { getOffsets =>
+      getOffsets(topicsToBeAssigned)


Good idea to restrict to the "new" ones.
Wouldn't topicPartitions or partitions be a better name?

Totally agree. Changed to partitions

ennru · 2018-05-23T13:21:13Z

core/src/main/scala/akka/kafka/internal/SubSourceLogic.scala

+      materializer.scheduleOnce(
+        settings.waitClosePartition,
+        new Runnable {
+          override def run(): Unit = cb.invoke(())


() => cb.invoke() should be enough.

ennru · 2018-05-23T13:21:37Z

core/src/test/scala/akka/kafka/scaladsl/IntegrationSpec.scala

@@ -332,7 +332,7 @@ class IntegrationSpec extends SpecBase(kafkaPort = KafkaPorts.IntegrationSpec) {
      }
    }

-    "call the onRevoked hook" in {
+    "call the onRevoked hook" ignore {


Why is this ignored?

This test was ignored because the new policy about re-use partitions. I've re-activated the test again. Now, we have to wait for the default time to ensure that the old partition is closed and a new one will be created.

ennru · 2018-05-23T14:28:28Z

Sorry, I misguided you - Scala 2.11 requires the Runnable.

dvallejo · 2018-05-23T14:43:01Z

No problem. I´ve reverted to Runnable

ennru

Congratulations! You're the first to experience the MigrationManager checks to fail.
We want to become serious on binary compatibility between versions and this tool helps us. The error indicated on Travis shows that introducing the parameter in the middle of things pushed a default parameter in the copy method around.

ennru · 2018-05-24T06:36:56Z

core/src/main/scala/akka/kafka/ConsumerSettings.scala

@@ -291,6 +292,7 @@ class ConsumerSettings[K, V](
    val closeTimeout: FiniteDuration,
    val commitTimeout: FiniteDuration,
    val wakeupTimeout: FiniteDuration,
+    val waitClosePartition: FiniteDuration,


Please move the new parameter last.

dvallejo · 2018-05-24T07:23:02Z

Oh, I was going crazy. I didn´t understand the fail 😄 Thanks!! I´ve changed the field to the last position.

ennru

A few more suggestions.
It might be worthwhile to sprinkle a few log messages into the logic, WDYT?

ennru · 2018-05-24T11:29:15Z

core/src/main/scala/akka/kafka/internal/SubSourceLogic.scala

              }
        }
    }
  }

+  var partitionsToRevoke: Set[TopicPartition] = Set.empty


Move these vars up to the other mutable fields, please.

ennru · 2018-05-24T11:30:56Z

core/src/main/scala/akka/kafka/internal/SubSourceLogic.scala

        .onComplete {
-          case Failure(ex) => stageFailCB.invoke(new ConsumerFailed(s"Failed to fetch offset for partitions: $tps.", ex))
+          case Failure(ex) => stageFailCB.invoke(new ConsumerFailed(s"Failed to fetch offset for partitions: $partitions.", ex))


Use ${partitions.mkString(", ")}

ennru · 2018-05-24T11:31:38Z

core/src/main/scala/akka/kafka/internal/SubSourceLogic.scala

              .recover {
-                case _: AskTimeoutException => stageFailCB.invoke(new ConsumerFailed(s"Consumer failed during seek for partitions: $tps."))
+                case _: AskTimeoutException => stageFailCB.invoke(new ConsumerFailed(s"Consumer failed during seek for partitions: $partitions."))


ennru · 2018-05-24T11:42:28Z

core/src/main/scala/akka/kafka/internal/SubSourceLogic.scala

              }
        }
    }
  }

+  var partitionsToRevoke: Set[TopicPartition] = Set.empty
+  var revokePendingCall: Option[Cancellable] = None


It is more like pendingRevokeCall.

ennru · 2018-05-24T11:45:59Z

core/src/main/scala/akka/kafka/internal/SubSourceLogic.scala

+      materializer.scheduleOnce(
+        settings.waitClosePartition,
+        new Runnable {
+          override def run(): Unit = cb.invoke(())


You could change revokePendingCall to None after invoking.

dvallejo · 2018-05-24T13:50:48Z

Totally agree. I've added some log messages and your requests. If you think there is not enough log messages, feel free to request more. Thanks for the review!

ennru

Some details...

ennru · 2018-05-25T12:49:50Z

core/src/main/scala/akka/kafka/internal/SubSourceLogic.scala

+    val partitions = tps -- partitionsToRevoke
+
+    if (partitions.nonEmpty) {
+      log.debug(s"Assigning new partitions: ${partitions.mkString(", ")}")


For logging you should use the log.debug("message {}", value) notation as it defers string creation.
In the particular case you might just add log.isDebugEnabled to the if clause.

ennru · 2018-05-25T12:52:05Z

core/src/main/scala/akka/kafka/internal/SubSourceLogic.scala

+      subSources --= partitionsToRevoke
+    }
+
+    log.debug(s"Waiting ${settings.waitClosePartition.toMillis} ms for pending requests before close partitions")


You might just use settings.waitClosePartition which will print its unit.

ennru

LGTM. I'll fix the logging.

ennru · 2018-05-29T12:33:25Z

Thank you for your contribution. Keep them coming!

edrevo · 2018-06-06T17:02:38Z

yay! many thanks @dvallejo for this PR! @ennru, do you know when 0.21 will be released? 😄

dvallejo added 6 commits May 23, 2018 11:07

fix: add-wait-in-rebalancing-to-close-partition

926afe5

chore: improve fix

730c8af

chore: ignore reassign topics in rebalancing

550551c

chore: add waitClosePartition to ConsumerSettings

8652f2e

ignore test because revoke is not always required

708e1c9

chore: remove internal test

97102bc

chore: remove unused imports

893d9c1

ennru reviewed May 23, 2018

View reviewed changes

chore: PR requests

46110cf

dvallejo force-pushed the fix-rebalances-with-partitions branch from 9708578 to 372821b Compare May 23, 2018 15:13

chore: revert to Runnable

350367e

dvallejo force-pushed the fix-rebalances-with-partitions branch from 372821b to 350367e Compare May 23, 2018 15:37

ennru reviewed May 24, 2018

View reviewed changes

chore: move property to the last one

f4ec78a

ennru reviewed May 24, 2018

View reviewed changes

chore: add logs and PR requests

d700ee2

ennru reviewed May 25, 2018

View reviewed changes

ennru approved these changes May 29, 2018

View reviewed changes

ennru merged commit b63df63 into akka:master May 29, 2018

ennru added this to the 0.21 milestone May 29, 2018

edrevo mentioned this pull request Jun 21, 2018

fix partition reassignment on partitioned sources #517

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Mitigate errors in rebalances with partitions in SubSourceLogic #466

Mitigate errors in rebalances with partitions in SubSourceLogic #466

dvallejo commented May 23, 2018

lightbend-cla-validator commented May 23, 2018

dvallejo commented May 23, 2018

ennru left a comment

ennru May 23, 2018

dvallejo May 23, 2018

ennru May 23, 2018

dvallejo May 23, 2018

ennru May 23, 2018

ennru May 23, 2018

dvallejo May 23, 2018 •

edited

Loading

ennru commented May 23, 2018

dvallejo commented May 23, 2018

ennru left a comment

ennru May 24, 2018

dvallejo commented May 24, 2018

ennru left a comment

ennru May 24, 2018

ennru May 24, 2018

ennru May 24, 2018

ennru May 24, 2018

ennru May 24, 2018

dvallejo commented May 24, 2018

ennru left a comment

ennru May 25, 2018

ennru May 25, 2018

ennru left a comment

ennru commented May 29, 2018

edrevo commented Jun 6, 2018

Mitigate errors in rebalances with partitions in SubSourceLogic #466

Mitigate errors in rebalances with partitions in SubSourceLogic #466

Conversation

dvallejo commented May 23, 2018

lightbend-cla-validator commented May 23, 2018

dvallejo commented May 23, 2018

ennru 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

dvallejo May 23, 2018 • edited Loading

Choose a reason for hiding this comment

ennru commented May 23, 2018

dvallejo commented May 23, 2018

ennru left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dvallejo commented May 24, 2018

ennru 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

dvallejo commented May 24, 2018

ennru left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ennru left a comment

Choose a reason for hiding this comment

ennru commented May 29, 2018

edrevo commented Jun 6, 2018

dvallejo May 23, 2018 •

edited

Loading