BUILD-KIP932: Commit to build branch with kip-932 librdkafka branch

Author: k-raina

.semaphore/semaphore.yml

@@ -9,6 +9,8 @@ global_job_config:
   env_vars:
     - name: LIBRDKAFKA_VERSION
       value: v2.13.2
+    - name: LIBRDKAFKA_BRANCH
+      value: dev_kip-932_queues-for-kafka
   prologue:
     commands:
       - checkout
@@ -332,7 +334,13 @@ blocks:
 
             # Build and install confluent-kafka from source
             - lib_dir=dest/runtimes/$OS_NAME-$ARCH/native
-            - tools/wheels/install-librdkafka.sh "${LIBRDKAFKA_VERSION#v}" dest
+            - |
+              if [[ -n $LIBRDKAFKA_BRANCH ]]; then
+                sudo apt-get install -y -qq libssl-dev libsasl2-dev liblz4-dev libzstd-dev
+                tools/wheels/build-librdkafka-branch.sh "$LIBRDKAFKA_BRANCH" dest
+              else
+                tools/wheels/install-librdkafka.sh "${LIBRDKAFKA_VERSION#v}" dest
+              fi
             - export CFLAGS="$CFLAGS -I${PWD}/dest/build/native/include"
             - export LDFLAGS="$LDFLAGS -L${PWD}/${lib_dir}"
             - export LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$PWD/$lib_dir"

tests/test_ShareConsumer.py

@@ -31,7 +31,7 @@ def test_constructor_with_valid_config():
     sc = ShareConsumer({
         'group.id': 'test-share-group',
         'bootstrap.servers': 'localhost:9092',
-        'session.timeout.ms': 100,
+        'socket.timeout.ms': 100,
     })
     assert sc is not None
     sc.close()
@@ -42,7 +42,7 @@ def test_subscribe():
     sc = ShareConsumer({
         'group.id': 'test-share-group',
         'bootstrap.servers': 'localhost:9092',
-        'session.timeout.ms': 100,
+        'socket.timeout.ms': 100,
     })
 
     sc.subscribe(['test-topic'])
@@ -59,7 +59,7 @@ def test_unsubscribe():
     sc = ShareConsumer({
         'group.id': 'test-share-group',
         'bootstrap.servers': 'localhost:9092',
-        'session.timeout.ms': 100,
+        'socket.timeout.ms': 100,
     })
 
     sc.subscribe(['test-topic'])
@@ -77,7 +77,7 @@ def test_consume_batch_no_broker():
         'group.id': 'test-share-group',
         'bootstrap.servers': 'localhost:9092',
         'socket.timeout.ms': '100',
-        'session.timeout.ms': 100,
+        'socket.timeout.ms': 100,
     })
 
     sc.subscribe(['test-topic'])
@@ -96,7 +96,7 @@ def test_close_idempotent():
     sc = ShareConsumer({
         'group.id': 'test-share-group',
         'bootstrap.servers': 'localhost:9092',
-        'session.timeout.ms': 100,
+        'socket.timeout.ms': 100,
     })
 
     sc.close()
@@ -108,7 +108,7 @@ def test_any_method_after_close_throws_exception():
     sc = ShareConsumer({
         'group.id': 'test-share-group',
         'bootstrap.servers': 'localhost:9092',
-        'session.timeout.ms': 100,
+        'socket.timeout.ms': 100,
     })
 
     sc.subscribe(['test-topic'])
@@ -148,7 +148,7 @@ def test_concurrent_consumers():
     kafka_config = {
         'group.id': 'test-share-group-integration',
         'bootstrap.servers': 'localhost:9092',
-        'session.timeout.ms': 100,
+        'socket.timeout.ms': 100,
     }
 
     sc1 = ShareConsumer(kafka_config)

tools/source-package-verification.sh

@@ -23,7 +23,11 @@ for version in 3.9.0 4.0.0; do
 done
 
 lib_dir=dest/runtimes/$OS_NAME-$ARCH/native
-tools/wheels/install-librdkafka.sh "${LIBRDKAFKA_VERSION#v}" dest
+if [[ -n $LIBRDKAFKA_BRANCH ]]; then
+    tools/wheels/build-librdkafka-branch.sh "$LIBRDKAFKA_BRANCH" dest
+else
+    tools/wheels/install-librdkafka.sh "${LIBRDKAFKA_VERSION#v}" dest
+fi
 export CFLAGS="$CFLAGS -I${PWD}/dest/build/native/include"
 export LDFLAGS="$LDFLAGS -L${PWD}/${lib_dir}"
 export LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$PWD/$lib_dir"

tools/wheels/build-librdkafka-branch.sh

@@ -0,0 +1,76 @@
+#!/bin/bash
+#
+# Build librdkafka from a git branch and install it into a NuGet-compatible
+# directory layout that matches what install-librdkafka.sh produces.
+#
+# Usage: build-librdkafka-branch.sh <branch> <destdir>
+#
+#   branch   - git branch name, e.g. dev_kip-932_queues-for-kafka
+#   destdir  - destination directory, e.g. dest
+#
+# Resulting layout (mirrors NuGet redist package):
+#   <destdir>/build/native/include/librdkafka/rdkafka.h
+#   <destdir>/runtimes/<os>-<arch>/native/librdkafka.{so.1,dylib}
+
+set -ex
+
+BRANCH="$1"
+DEST="$2"
+
+if [[ -z $BRANCH || -z $DEST ]]; then
+    echo "Usage: $0 <branch> <destdir>"
+    exit 1
+fi
+
+if [[ -f $DEST/build/native/include/librdkafka/rdkafka.h ]]; then
+    echo "$0: librdkafka already built in $DEST"
+    exit 0
+fi
+
+echo "$0: Building librdkafka branch '$BRANCH' into '$DEST'"
+
+ARCH=${ARCH:-x64}
+SRC=/tmp/librdkafka-branch-src
+INSTALL=$SRC/install
+
+[[ -d "$DEST" ]] || mkdir -p "$DEST"
+rm -rf "$SRC"
+
+git clone --depth 1 --branch "$BRANCH" \
+    https://github.com/confluentinc/librdkafka.git "$SRC"
+
+pushd "$SRC"
+./configure --prefix="$INSTALL" --disable-debug-symbols
+make -j"$(nproc 2>/dev/null || sysctl -n hw.ncpu)"
+make install
+popd
+
+# --- Mirror NuGet layout ---
+
+INC_DST="$DEST/build/native/include"
+mkdir -p "$INC_DST"
+cp -r "$INSTALL/include/librdkafka" "$INC_DST/"
+
+if [[ $OSTYPE == linux* ]]; then
+    OS_NAME=linux
+    LIB_DST="$DEST/runtimes/$OS_NAME-$ARCH/native"
+    mkdir -p "$LIB_DST"
+    cp -v "$INSTALL"/lib/librdkafka.so* "$LIB_DST/" 2>/dev/null || true
+    # Ensure librdkafka.so.1 exists (needed by the loader)
+    if [[ ! -f "$LIB_DST/librdkafka.so.1" ]]; then
+        cp -v "$LIB_DST/librdkafka.so" "$LIB_DST/librdkafka.so.1"
+    fi
+    ldd "$LIB_DST/librdkafka.so.1"
+
+elif [[ $OSTYPE == darwin* ]]; then
+    OS_NAME=osx
+    LIB_DST="$DEST/runtimes/$OS_NAME-$ARCH/native"
+    mkdir -p "$LIB_DST"
+    cp -v "$INSTALL"/lib/librdkafka*.dylib "$LIB_DST/"
+    # Fix the dylib self-referencing name to its installed path
+    install_name_tool -id "$LIB_DST/librdkafka.dylib" "$LIB_DST/librdkafka.dylib"
+    otool -L "$LIB_DST/librdkafka.dylib"
+fi
+
+rm -rf "$SRC"
+echo "$0: Done. Headers at $INC_DST, library at $LIB_DST"