Repository: beatlabs/patron
Pull request #346 (2021-02-11 09:16:27)

Kafka Duration Client: Time Extractor Failure
Signed-off-by: teivah <t.harsanyi@thebeat.co>

## Which problem is this PR solving?

Closes https://github.com/beatlabs/patron/issues/345

## Short description of the changes

* Log a warning if the time extractor fails
* Update the range of the binary search so that we don't keep consuming the same message


codecov[bot]: # [Codecov](https://codecov.io/gh/beatlabs/patron/pull/346?src=pr&el=h1) Report
> Merging [#346](https://codecov.io/gh/beatlabs/patron/pull/346?src=pr&el=desc) (1aa9e6a) into [master](https://codecov.io/gh/beatlabs/patron/commit/8ca918b74ae13960206696d9f1db6070db62dd13?el=desc) (8ca918b) will **decrease** coverage by `0.04%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/beatlabs/patron/pull/346/graphs/tree.svg?width=650&height=150&src=pr&token=sxY15rXW1X)](https://codecov.io/gh/beatlabs/patron/pull/346?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master     #346      +/-   ##
==========================================
- Coverage   74.36%   74.32%   -0.05%     
==========================================
  Files          65       65              
  Lines        3678     3680       +2     
==========================================
  Hits         2735     2735              
- Misses        838      839       +1     
- Partials      105      106       +1     
```


| [Impacted Files](https://codecov.io/gh/beatlabs/patron/pull/346?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [component/async/kafka/simple/duration\_client.go](https://codecov.io/gh/beatlabs/patron/pull/346/diff?src=pr&el=tree#diff-Y29tcG9uZW50L2FzeW5jL2thZmthL3NpbXBsZS9kdXJhdGlvbl9jbGllbnQuZ28=) | `96.49% <100.00%> (+0.12%)` | :arrow_up: |
| [component/sqs/component.go](https://codecov.io/gh/beatlabs/patron/pull/346/diff?src=pr&el=tree#diff-Y29tcG9uZW50L3Nxcy9jb21wb25lbnQuZ28=) | `86.66% <0.00%> (-1.12%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/beatlabs/patron/pull/346?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/beatlabs/patron/pull/346?src=pr&el=footer). Last update [8ca918b...b38ac8c](https://codecov.io/gh/beatlabs/patron/pull/346?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/component/async/kafka/simple/duration_client.go b/component/async/kafka/simple/duration_client.go
index ec3366d44..2636e3888 100644
--- a/component/async/kafka/simple/duration_client.go
+++ b/component/async/kafka/simple/duration_client.go
@@ -112,7 +112,10 @@ func (d durationClient) offsetBinarySearch(ctx context.Context, topic string, si
 
 		t, err := timeExtractor(msg)
 		if err != nil {
-			return 0, fmt.Errorf("error while executing comparator: %w", err)
+			log.FromContext(ctx).Warnf("error while executing comparator: %v", err)
+			// In case of a failure, we compress the range so that the next calculated mid is different
+			left++
+			continue
 		}
 
 		if t.Equal(since) {


Test patch:
diff --git a/component/async/kafka/simple/duration_client_test.go b/component/async/kafka/simple/duration_client_test.go
index 67dedbbe0..e761358f6 100644
--- a/component/async/kafka/simple/duration_client_test.go
+++ b/component/async/kafka/simple/duration_client_test.go
@@ -117,6 +117,35 @@ func Test_Consumer_GetTimeBasedOffsetsPerPartition(t *testing.T) {
 				0: 0,
 			},
 		},
+		"success - with invalid message": {
+			globalTimeout: time.Second,
+			client: client(topic).
+				partitionIDs([]int32{0}, nil).
+				partition(0, partitionConfig{
+					oldest: offset{offset: 0},
+					newest: offset{offset: 10},
+					messages: map[int64]messageAtOffset{
+						0: {
+							msg: &sarama.ConsumerMessage{Timestamp: since.Add(1 * time.Hour)},
+						},
+						1: {
+							msg: invalidMessage,
+						},
+						2: {
+							msg: &sarama.ConsumerMessage{Timestamp: since.Add(3 * time.Hour)},
+						},
+						3: {
+							msg: &sarama.ConsumerMessage{Timestamp: since.Add(4 * time.Hour)},
+						},
+						4: {
+							msg: &sarama.ConsumerMessage{Timestamp: since.Add(5 * time.Hour)},
+						},
+					},
+				}).build(),
+			expectedOffsets: map[int32]int64{
+				0: 2,
+			},
+		},
 		"success - all outside": {
 			globalTimeout: time.Second,
 			client: client(topic).
@@ -214,21 +243,6 @@ func Test_Consumer_GetTimeBasedOffsetsPerPartition(t *testing.T) {
 				}).build(),
 			expectedErr: errors.New("foo"),
 		},
-		"error - invalid message": {
-			globalTimeout: time.Second,
-			client: client(topic).
-				partitionIDs([]int32{0}, nil).
-				partition(0, partitionConfig{
-					oldest: offset{offset: 0},
-					newest: offset{offset: 10},
-					messages: map[int64]messageAtOffset{
-						4: {
-							msg: invalidMessage,
-						},
-					},
-				}).build(),
-			expectedErr: errors.New("error while executing comparator: empty time"),
-		},
 		"error - get message offset": {
 			globalTimeout: time.Second,
 			client: client(topic).