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

Jump to bottom

Fix bad offset management #597

Merged

knative-prow-robot merged 5 commits into knative-extensions:master from slinkydeveloper:issues/380

Feb 8, 2021

Contributor

slinkydeveloper commented Feb 4, 2021

Signed-off-by: Francesco Guardiani francescoguard@gmail.com

Fixes #380

Proposed Changes

🐛 Now offset management is correct

Release Note

:bug: Now offset management is correct

Docs


          Fix knative-extensions#380

4466f58

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

google-cla bot added the cla: yes label

knative-prow-robot added the area/data-plane label

knative-prow-robot requested review from Harwayne and pierDipi

February 4, 2021 17:16

knative-prow-robot added approved size/L labels

codecov bot commented Feb 4, 2021 •

edited

Loading

Codecov Report

Merging #597 (e770db8) into master (482f1c3) will decrease coverage by 8.23%.
The diff coverage is 100.00%.

@@             Coverage Diff              @@
##             master     #597      +/-   ##
============================================
- Coverage     71.85%   63.62%   -8.24%     
============================================
  Files            72       28      -44     
  Lines          2587     1207    -1380     
  Branches        121        0     -121     
============================================
- Hits           1859      768    -1091     
+ Misses          584      364     -220     
+ Partials        144       75      -69

Flag	Coverage Δ	Complexity Δ
java-unittests	`?`	`?`

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ	Complexity Δ
control-plane/pkg/core/config/utils.go	`72.41% <100.00%> (+0.98%)`	`0.00 <0.00> (ø)`
control-plane/pkg/reconciler/broker/broker.go	`78.71% <100.00%> (ø)`	`0.00 <0.00> (ø)`
control-plane/pkg/reconciler/sink/kafka_sink.go	`75.42% <100.00%> (ø)`	`0.00 <0.00> (ø)`
control-plane/pkg/reconciler/trigger/trigger.go	`71.66% <100.00%> (-0.24%)`	`0.00 <0.00> (ø)`
...fka/broker/core/security/PlaintextCredentials.java
...a/broker/core/tracing/HeadersPropagatorSetter.java
...a/broker/core/security/KubernetesAuthProvider.java
.../kafka/broker/dispatcher/ConsumerRecordSender.java
...ka/broker/core/security/KubernetesCredentials.java
...nting/kafka/broker/core/tracing/TracingConfig.java
... and 29 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 482f1c3...c8da4e4. Read the comment docs.

Contributor Author

slinkydeveloper commented Feb 4, 2021

/retest

1 similar comment

Contributor Author

slinkydeveloper commented Feb 4, 2021

/retest

slinkydeveloper commented

View reviewed changes

...knative/eventing/kafka/broker/dispatcher/strategy/UnorderedConsumerRecordOffsetStrategy.java Outdated Show resolved Hide resolved

pierDipi reviewed

View reviewed changes

...ive/eventing/kafka/broker/dispatcher/strategy/UnorderedConsumerRecordOffsetStrategyTest.java Outdated

Comment on lines 235 to 236

		// If you don't want to lose hours in debugging, please don't remove this FQCNs :)
		final Map<io.vertx.kafka.client.common.TopicPartition, io.vertx.kafka.client.consumer.OffsetAndMetadata>

Member

pierDipi Feb 4, 2021

LOL 🤣

...knative/eventing/kafka/broker/dispatcher/strategy/UnorderedConsumerRecordOffsetStrategy.java Outdated Show resolved Hide resolved

slinkydeveloper mentioned this pull request

Ordered consumer #589

Closed

7 tasks

Fix

8d7bb5f

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

pierDipi reviewed

View reviewed changes

...knative/eventing/kafka/broker/dispatcher/strategy/UnorderedConsumerRecordOffsetStrategy.java Outdated Show resolved Hide resolved

...knative/eventing/kafka/broker/dispatcher/strategy/UnorderedConsumerRecordOffsetStrategy.java Outdated Show resolved Hide resolved

...knative/eventing/kafka/broker/dispatcher/strategy/UnorderedConsumerRecordOffsetStrategy.java Outdated Show resolved Hide resolved

...knative/eventing/kafka/broker/dispatcher/strategy/UnorderedConsumerRecordOffsetStrategy.java

+                    consumer.commit(Map.of(
+                      key,
+                      new OffsetAndMetadata(shouldAck + 1, ""))
+                    ).onSuccess(ignored -> {

Member

pierDipi Feb 8, 2021

onSuccess on a new line?

Contributor Author

slinkydeveloper Feb 8, 2021

The closed paren is right there 😄

...knative/eventing/kafka/broker/dispatcher/strategy/UnorderedConsumerRecordOffsetStrategy.java Outdated Show resolved Hide resolved

...knative/eventing/kafka/broker/dispatcher/strategy/UnorderedConsumerRecordOffsetStrategy.java Outdated Show resolved Hide resolved

...knative/eventing/kafka/broker/dispatcher/strategy/UnorderedConsumerRecordOffsetStrategy.java Outdated Show resolved Hide resolved

...knative/eventing/kafka/broker/dispatcher/strategy/UnorderedConsumerRecordOffsetStrategy.java Outdated

+                        // If the commit failed, there are 2 situations:
+                        // * Nobody tried to commit again, so let's just restore the state
+                        // * Somebody committed with an offset greater than this one, so we just discard the error
+                        if (!(this.lastAckedMap.get(key) > shouldAck)) {

Member

pierDipi Feb 8, 2021

Suggested change

      
                      if (!(this.lastAckedMap.get(key) > shouldAck)) {
          
                      if (this.lastAckedMap.get(key) <= shouldAck) {

Contributor Author

slinkydeveloper Feb 8, 2021 •

edited

Loading

I would love to keep it this way, since it's consistent with the above comment

...knative/eventing/kafka/broker/dispatcher/strategy/UnorderedConsumerRecordOffsetStrategy.java Outdated Show resolved Hide resolved

...knative/eventing/kafka/broker/dispatcher/strategy/UnorderedConsumerRecordOffsetStrategy.java Outdated

Comment on lines 125 to 126

		long lastAckedBeforeThisOne = this.lastAckedMap.put(key, shouldAck);
		SortedSet<Long> messagesImGoingToAck = this.pendingAcksMap.remove(key);

Member

pierDipi Feb 8, 2021

This should be updated on success or on failure.

Contributor Author

slinkydeveloper Feb 8, 2021

No, this is discarded in case of success, copy pasted in case of failure (if we actually need to restore the state)

slinkydeveloper added 2 commits

February 8, 2021 17:14


          Comments

5949bd5

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>


          Comments

aee5633

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

pierDipi reviewed

View reviewed changes

...knative/eventing/kafka/broker/dispatcher/strategy/UnorderedConsumerRecordOffsetStrategy.java Outdated

+                        if (!(this.lastAckedPerPartition.get(topicPartition) > toAck)) {
+                          this.lastAckedPerPartition.put(topicPartition, lastAckedBeforeThisOne);
+                          this.pendingAcksPerPartition.compute(topicPartition, (k, actual) -> {
+                            if (actual != null) {

Member

pierDipi Feb 8, 2021

Is this possible since we call mutateStateAndCheckAck above?

Contributor Author

slinkydeveloper Feb 8, 2021

It is, because somebody could have asked meanwhile to add a new commit and that state could have been populated again

...knative/eventing/kafka/broker/dispatcher/strategy/UnorderedConsumerRecordOffsetStrategy.java Outdated

Comment on lines 155 to 164

+                        if (!(this.lastAckedPerPartition.get(topicPartition) > toAck)) {
+                          this.lastAckedPerPartition.put(topicPartition, lastAckedBeforeThisOne);
+                          this.pendingAcksPerPartition.compute(topicPartition, (k, actual) -> {
+                            if (actual != null) {
+                              actual.addAll(messagesImGoingToAck);
+                              return actual;
+                            }
+                            return messagesImGoingToAck;
+                          });
+                        }

Member

pierDipi Feb 8, 2021

If I comment out these lines, tests are green.

Member

pierDipi Feb 8, 2021

Also, if we set the state on success, there is no need to rollback.

Member

pierDipi Feb 8, 2021

It should be safer since the state could have changed since lastAckedBeforeThisOne was calculated.

Contributor Author

slinkydeveloper Feb 8, 2021

If I comment out these lines, tests are green.

How could it be? Lemme check 🤔

Contributor Author

slinkydeveloper Feb 8, 2021 •

edited

Loading

Ah yes of course it goes green, but if the system crashes after trying to commit, it will restart from the previous ack. Do we care? Otherwise I'll just remove that statement, it just makes the code more complicated


          Removed complicated code which was practically useless

c8da4e4

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

pierDipi approved these changes

View reviewed changes

Member

pierDipi left a comment

/lgtm

Thank you!

/kind bug

knative-prow-robot added the kind/bug label

knative-prow-robot assigned pierDipi

knative-prow-robot added the lgtm label

Contributor

knative-prow-robot commented Feb 8, 2021

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: pierDipi, slinkydeveloper

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Needs approval from an approver in each of these files:

~~OWNERS~~ [pierDipi,slinkydeveloper]

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

Member

pierDipi commented Feb 8, 2021

Java tests failed with Connection reset in GH.

knative-prow-robot merged commit f584a4e into knative-extensions:master

pierDipi pushed a commit to pierDipi/eventing-kafka-broker that referenced this pull request


          Fix bad offset management (knative-extensions#597)

9606c86

* Fix knative-extensions#380

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Fix

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Comments

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Comments

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Removed complicated code which was practically useless

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

matzew pushed a commit to matzew/eventing-kafka-broker that referenced this pull request


          [release-v1.7] [SRVKE-1400] Autoscaler improvements (knative-extensio…

9624ffe

…ns#597)

* Periodically resync consumer groups to downscale statefulset

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

* Autoscaler scale down delay

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

* Scale machineset for rekt tests

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

* Increase is ready timeout

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

* Skip namespace cleanup so that we can more must gather info

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

---------

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>
Co-authored-by: Pierangelo Di Pilato <pierdipi@redhat.com>

This pull request was closed.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

approved area/data-plane cla: yes kind/bug lgtm size/L