KAFKA-9966: add internal assignment listener to stabilize eos-beta upgrade test by ableegoldman · Pull Request #8648 · apache/kafka

ableegoldman · 2020-05-12T02:07:06Z

Adds an internal assignment listener interface with a callback that exposes the assignment stability. A useful implementation that tracks the number of stable assignments is available in IntegrationTestUtils, and leveraged to fix the flaky EOSBetaUpgradeTest, but any implementation can be plugged in to meet the specific needs of a test.

ableegoldman · 2020-05-12T02:10:41Z

Will need to be rebased once pull/8596 is merged, but the PR is ready. Reduced the nearly 100% failure rate of this test on top of the changes in pull/8596 to 0% (that I've seen so far 🙂) @mjsax @vvcephei

edit: this PR alone does not contain all the fixes needed for the eos-beta upgrade test, so I'm leaving it ignored for now. The other set of fixes are included in #8662

We can remove the Ignore annotation in whichever PR gets merged last

mjsax · 2020-05-12T21:45:08Z

Why do we use getLogPrefix() -- our code style always omit the get prefix -- should be just logPrefix() (Similar below for other methods)

I guess I thought there was a reason getXX was used in this class and figured all the methods should conform. Idk. I'll remove the get

ableegoldman · 2020-05-12T22:50:07Z

        }
    }

+    public AssignmentListener assignmentListener() {


This is the only non-cosmetic change in this class, along with the interface added below

ableegoldman · 2020-05-12T22:53:16Z

-    private final org.apache.kafka.streams.processor.PartitionGrouper partitionGrouper;
-    private final String userEndPoint;
    private final TaskManager taskManager;
-    private final StreamsMetadataState streamsMetadataState;


Sorry for all the lines of changes here, but the inconsistent method signatures and object construction was making this class hard to follow. All I'm doing here is making the methods conform to the same style, and moving the construction of any config that isn't needed elsewhere to its getter method. This allows us to remove a lot of these class variables

Do we know if the corresponding "getters" are called often or not? I guess the main idea behind having those member variables was to "parse" the config once as it's immutable anyway instead of each time a getter is called?

They are not. We construct the AssignorConfiguration object once in StreamsPartitionAssignor#configure and call the getters immediately afterwards, and only there

ableegoldman · 2020-05-12T22:55:41Z

-                                        final List<KeyValue<KafkaStreams.State, KafkaStreams.State>> expected)
-            throws Exception {
-
+    private void waitForRunning(final List<KeyValue<KafkaStreams.State, KafkaStreams.State>> observed) throws Exception {


This one actually might not be necessary, but I thought it was useful as a sanity check especially since we're trying to access the stores in some places

mjsax

Overall LGTM.

mjsax · 2020-05-12T23:30:04Z

-    private final org.apache.kafka.streams.processor.PartitionGrouper partitionGrouper;
-    private final String userEndPoint;
    private final TaskManager taskManager;
-    private final StreamsMetadataState streamsMetadataState;


Do we know if the corresponding "getters" are called often or not? I guess the main idea behind having those member variables was to "parse" the config once as it's immutable anyway instead of each time a getter is called?

mjsax · 2020-05-12T23:33:51Z

-                                        final List<KeyValue<KafkaStreams.State, KafkaStreams.State>> expected)
-            throws Exception {
-
+    private void waitForRunning(final List<KeyValue<KafkaStreams.State, KafkaStreams.State>> observed) throws Exception {


mjsax · 2020-05-12T23:34:34Z

+    private void waitForRunning(final List<KeyValue<KafkaStreams.State, KafkaStreams.State>> observed) throws Exception {
        waitForCondition(
-            () -> observed.equals(expected),
+            () -> observed.get(observed.size() - 1).equals(new KeyValue<>(State.REBALANCING, State.RUNNING)),


Seems this could fail if observed is empty?

Also, could we simplify this? We are only interested in RUNNING state, right? So no need to check the previous state?

True, I'll fix this

mjsax · 2020-05-12T23:36:24Z


    @Test
    public void shouldThrowExceptionIfApplicationServerConfigPortIsNotAnInteger() {
+        createDefaultMockTaskManager();


Why this change?

If we don't configure the TaskManager then we'll throw (the wrong kind of) exception during configure. Previously this test just happened to pass because we processed the user endpoint before we processed the TaskManager, so we never got to the TaskManager

mjsax · 2020-05-12T23:39:03Z

-                    KeyValue.pair(KafkaStreams.State.REBALANCING, KafkaStreams.State.RUNNING)
-                )
-            );
+            assignmentListener.waitForNextStableAssignment(expectedNumStableAssignments, MAX_WAIT_TIME_MS);


So far we always pass in assignmentListener.numStableAssignments() + 1 -- hence, I am wondering if we need to pass it at all?

Well we need to record it before doing whatever action causes the group to rebalance. But maybe we can do something like prepareForRebalance if that seems easier?

I see. Guess it's good as-is.

mjsax · 2020-05-12T23:42:21Z

Retest this please.

mjsax

LGTM.

mjsax · 2020-05-13T00:26:46Z

Retest this please.

mjsax · 2020-05-13T00:27:27Z

Retest this please.

ableegoldman · 2020-05-13T04:11:05Z

Java 14 passed, Java 8 build failed due to mysterious 17:27:43 ERROR: Error cloning remote repo 'origin' and Java 11 failed due to flaky MirrorConnectorsIntegrationTest.testReplication (reported on ticket)

* 'trunk' of github.com:apache/kafka: MINOR: add option to rebuild source for system tests (apache#6656) KAFKA-9850 Move KStream#repartition operator validation during Topolo… (apache#8550) MINOR: Add a duplicate() method to Message classes (apache#8556) KAFKA-9966: add internal assignment listener to stabilize eos-beta upgrade test (apache#8648) MINOR: Replace null with an actual value for timestamp field in InsertField SMT unit tests (apache#8649) MINOR: Fix ProcessorContext JavaDocs and stream-time computation (apache#8603) MINOR: improve tests for TopologyTestDriver (apache#8631) KAFKA-9821: consolidate Streams rebalance triggering mechanisms (apache#8596) KAFKA-9669; Loosen validation of inner offsets for older message formats (apache#8647) KAFKA-8770: KIP-557: Drop idempotent KTable source updates (apache#8254) MINOR: Remove allow concurrent test (apache#8641)

…grade test (apache#8648) Reviewer: Matthias J. Sax <matthias@confluent.io>

ableegoldman force-pushed the MINOR-assignment-callback-for-integration-tests branch from 58f5eb2 to c033c3c Compare May 12, 2020 19:36

mjsax reviewed May 12, 2020

View reviewed changes

ableegoldman force-pushed the MINOR-assignment-callback-for-integration-tests branch from c033c3c to 276aae8 Compare May 12, 2020 22:32

refactor AssignorConfiguration some, add listener for callback

37bf2a6

ableegoldman force-pushed the MINOR-assignment-callback-for-integration-tests branch from 276aae8 to 37bf2a6 Compare May 12, 2020 22:34

ableegoldman added 3 commits May 12, 2020 15:37

remove get prefix

21e51d2

keep Ignore for now

3654b47

checkstyle

c4c3ed6

ableegoldman commented May 12, 2020

View reviewed changes

mjsax reviewed May 12, 2020

View reviewed changes

mjsax added streams tests Test fixes (including flaky tests) labels May 12, 2020

github review changes

959e221

mjsax approved these changes May 13, 2020

View reviewed changes

mjsax merged commit 53875bb into apache:trunk May 13, 2020

jwijgerd pushed a commit to buxapp/kafka that referenced this pull request May 14, 2020

KAFKA-9966: add internal assignment listener to stabilize eos-beta up…

360be0d

…grade test (apache#8648) Reviewer: Matthias J. Sax <matthias@confluent.io>

ableegoldman deleted the MINOR-assignment-callback-for-integration-tests branch May 15, 2020 01:03

Uh oh!

Conversation

ableegoldman commented May 12, 2020

Uh oh!

ableegoldman commented May 12, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

mjsax commented May 12, 2020

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

mjsax commented May 13, 2020

Uh oh!

mjsax commented May 13, 2020

Uh oh!

ableegoldman commented May 13, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

ableegoldman commented May 12, 2020 •

edited

Loading