Merge pull request #22 from redBorder/development

Release 1.1.7

.github/workflows/rpm.yml

@@ -0,0 +1,66 @@
+name: RPM Build and Upload
+
+on:
+  push:
+    branches:
+      - 'master'
+      - 'main'
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+
+    env:
+      ACTIONS_ALLOW_UNSECURE_COMMANDS: true
+
+    steps:
+      - name: Checkout Repository
+        uses: actions/checkout@v2
+        with:
+          fetch-depth: 0
+
+      - name: Create tag based on VERSION
+        id: create_tag
+        run: |
+          TAG=$(cat ./VERSION)
+          echo "TAG=$TAG" >> $GITHUB_ENV
+        shell: bash
+
+      - name: Set Version
+        run: echo "VERSION=${{ env.TAG }}" >> $GITHUB_ENV
+
+      - name: Run Docker Container
+        run: docker run --privileged -d --name builder --network host rockylinux:9 /bin/sleep infinity
+
+      - name: Install build tools RPM
+        run: |
+          docker cp ./ builder:/build
+          docker exec builder bash -c "yum install -y epel-release && yum install -y make git mock"
+          docker exec builder bash -c "rm -rf /etc/mock/default.cfg"
+
+      - name: Setup SDK
+        run: |
+          docker exec builder bash -c "curl https://raw.githubusercontent.com/redBorder/repoinit/master/sdk9.cfg > /build/sdk9.cfg"
+          docker exec builder bash -c "echo \"config_opts['use_host_resolv'] = True\" >> /build/sdk9.cfg"
+          docker exec builder bash -c "ln -s /build/sdk9.cfg /etc/mock/default.cfg"
+
+      - name: Build RPM using mock
+        run: |
+          docker exec builder bash -c "git config --global --add safe.directory /build"
+          docker exec builder bash -c "cd /build/ && VERSION=${{ env.TAG }} make rpm"
+
+      - name: Copy RPMS
+        run: |
+          docker cp builder:/build/packaging/rpm/pkgs/. ./rpms
+
+      - name: Delete non-.rpm files
+        run: |
+          find ./rpms -type f -not -name '*.rpm' -exec rm {} \;
+
+      - name: Release
+        uses: softprops/action-gh-release@v1
+        with:
+          files: ./rpms/*
+          tag_name: ${{ env.TAG }}
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

VERSION

@@ -0,0 +1 @@
+1.1.7

kafka.go

@@ -19,13 +19,14 @@ package main
 import (
 	"os"
 	"os/signal"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
 
 	"github.com/IBM/sarama"
 	"github.com/redBorder/rbforwarder"
-	"github.com/redBorder/sarama-cluster"
+	cluster "github.com/redBorder/sarama-cluster"
 	"github.com/sirupsen/logrus"
 )
 
@@ -43,6 +44,7 @@ type KafkaConsumer struct {
 	forwarder *rbforwarder.RBForwarder // The backend to send messages
 	consumer  *cluster.Consumer
 	closed    chan struct{}
+	reset     chan struct{}
 	Config    KafkaConfig // Cofiguration after the parsing
 	wg        sync.WaitGroup
 }
@@ -55,6 +57,7 @@ func (k *KafkaConsumer) Start() {
 	var err error
 
 	k.closed = make(chan struct{})
+	k.reset = make(chan struct{})
 
 	logger = Logger.WithFields(logrus.Fields{
 		"prefix": "k2http",
@@ -101,6 +104,8 @@ consumerLoop:
 			k.Config.topics,
 			k.Config.consumerGroupConfig,
 		)
+		k.Config.consumerGroupConfig.Consumer.Return.Errors = true
+
 		if err != nil {
 			logger.Fatal("Failed to start consumer: ", err)
 			break
@@ -112,17 +117,34 @@ consumerLoop:
 			WithField("topics", k.Config.topics).
 			Info("Started consumer")
 
+		go func() {
+			for err := range k.consumer.Errors() {
+				if strings.Contains(err.Error(), "offset is outside the range") {
+					logger.Infof("Resetting offsets due to out of range error")
+					k.reset <- struct{}{}
+					return
+				}
+			}
+		}()
+
 		for {
 			select {
 			case <-k.closed:
 				break consumerLoop
+			case <-k.reset:
+				logger.Infof("Reset signal received, closing consumer to reset offsets")
+				if err := k.consumer.Close(); err != nil {
+					logger.Error("Failed to close consumer: ", err)
+				}
+				k.Config.consumerGroupConfig.Consumer.Offsets.Initial = sarama.OffsetOldest
+				continue consumerLoop
 			case message := <-k.consumer.Messages():
 				if message == nil {
 					k.Config.consumerGroupConfig.Consumer.Offsets.Initial = sarama.OffsetOldest
 					if err := k.consumer.Close(); err != nil {
 						logger.Error("Failed to close consumer: ", err)
 					}
-
+					logger.Error("Message is nill")
 					break
 				}
 

packaging/rpm/k2http.spec

@@ -64,7 +64,7 @@ exit 0
 /usr/lib/systemd/system/k2http.service
 
 %changelog
-* Tue Apr 02 2024 David Vanhoucke <dvanhoucke@redborder.com> - 1.1.6
+* Tue Apr 02 2024 David Vanhoucke <dvanhoucke@redborder.com> - 1.1.7
 - Dependencies changed, erased the building of librdkafka from git repo and no longer giving permissions to config.yml
 * Tue Feb 08 2022 Vicente Mesa <vimesa@redborder.com> - 1.0.0
 - first spec version