From 5dd07815e9f95ac4ff906b923b0cc82e8c8fdb73 Mon Sep 17 00:00:00 2001
From: Frank Hamand <frankhamand@gmail.com>
Date: Tue, 23 Jul 2024 15:51:33 +0100
Subject: [PATCH] fix: Add epoch leader bug fix to kafka patch (#23852)

---
 .../patches/node-rdkafka@2.17.0.patch         | 46 +++++++++++++++++++
 plugin-server/pnpm-lock.yaml                  |  6 +--
 2 files changed, 49 insertions(+), 3 deletions(-)

diff --git a/plugin-server/patches/node-rdkafka@2.17.0.patch b/plugin-server/patches/node-rdkafka@2.17.0.patch
index 20d8c5c8438..a83bf425bd5 100644
--- a/plugin-server/patches/node-rdkafka@2.17.0.patch
+++ b/plugin-server/patches/node-rdkafka@2.17.0.patch
@@ -1079,3 +1079,49 @@ index 0f4de520ed6b8a06dfe355e0bb9091273def98a5..ada72a7e621ea5433f194ab3d22eef32
  var t = require('assert');
  
  var client;
+diff --git a/deps/librdkafka/src/rdkafka_partition.h b/deps/librdkafka/src/rdkafka_partition.h
+index f9dd686423..aef704b95f 100644
+--- a/deps/librdkafka/src/rdkafka_partition.h
++++ b/deps/librdkafka/src/rdkafka_partition.h
+@@ -68,24 +68,35 @@ struct rd_kafka_toppar_err {
+                                   *   last msg sequence */
+ };
+ 
+-
++/**
++ * @brief Fetchpos comparator, only offset is compared.
++ */
++static RD_UNUSED RD_INLINE int
++rd_kafka_fetch_pos_cmp_offset(const rd_kafka_fetch_pos_t *a,
++                              const rd_kafka_fetch_pos_t *b) {
++        if (a->offset < b->offset)
++                return -1;
++        else if (a->offset > b->offset)
++                return 1;
++        else
++                return 0;
++}
+ 
+ /**
+  * @brief Fetchpos comparator, leader epoch has precedence.
++ *        iff both values are not null.
+  */
+ static RD_UNUSED RD_INLINE int
+ rd_kafka_fetch_pos_cmp(const rd_kafka_fetch_pos_t *a,
+                        const rd_kafka_fetch_pos_t *b) {
++        if (a->leader_epoch == -1 || b->leader_epoch == -1)
++                return rd_kafka_fetch_pos_cmp_offset(a, b);
+         if (a->leader_epoch < b->leader_epoch)
+                 return -1;
+         else if (a->leader_epoch > b->leader_epoch)
+                 return 1;
+-        else if (a->offset < b->offset)
+-                return -1;
+-        else if (a->offset > b->offset)
+-                return 1;
+         else
+-                return 0;
++                return rd_kafka_fetch_pos_cmp_offset(a, b);
+ }
+ 
+ 
diff --git a/plugin-server/pnpm-lock.yaml b/plugin-server/pnpm-lock.yaml
index e966fdd3e01..ea69c6b1b7c 100644
--- a/plugin-server/pnpm-lock.yaml
+++ b/plugin-server/pnpm-lock.yaml
@@ -6,7 +6,7 @@ settings:
 
 patchedDependencies:
   node-rdkafka@2.17.0:
-    hash: p4aetcvho53cvjti6c3zi7tfaq
+    hash: bugorwxdhlhl2utknko2c5ibqm
     path: patches/node-rdkafka@2.17.0.patch
 
 dependencies:
@@ -114,7 +114,7 @@ dependencies:
     version: 2.6.9
   node-rdkafka:
     specifier: ^2.17.0
-    version: 2.17.0(patch_hash=p4aetcvho53cvjti6c3zi7tfaq)
+    version: 2.17.0(patch_hash=bugorwxdhlhl2utknko2c5ibqm)
   node-schedule:
     specifier: ^2.1.0
     version: 2.1.1
@@ -8431,7 +8431,7 @@ packages:
     resolution: {integrity: sha512-O5lz91xSOeoXP6DulyHfllpq+Eg00MWitZIbtPfoSEvqIHdl5gfcY6hYzDWnj0qD5tz52PI08u9qUvSVeUBeHw==}
     dev: true
 
-  /node-rdkafka@2.17.0(patch_hash=p4aetcvho53cvjti6c3zi7tfaq):
+  /node-rdkafka@2.17.0(patch_hash=bugorwxdhlhl2utknko2c5ibqm):
     resolution: {integrity: sha512-vFABzRcE5FaH0WqfqJRxDoqeG6P8UEB3M4qFQ7SkwMgQueMMO78+fm8MYfl5hLW3bBYfBekK2BXIIr0lDQtSEQ==}
     engines: {node: '>=6.0.0'}
     requiresBuild: true