added version check before starting reconciliation

adapters/backend/v1/adapter.go

@@ -219,6 +219,14 @@ func (a *Adapter) startReconciliationPeriodicTask(mainCtx context.Context, cfg *
 						logger.L().Error("expected to find client for reconciliation in clients map", helpers.String("clientId", clientId.String()))
 						continue
 					}
+
+					if !utils.IsBatchMessageSupported(clientId.Version) {
+						logger.L().Info("skipping reconciliation request for client because it does not support batch messages",
+							helpers.String("version", clientId.Version),
+							helpers.Interface("clientId", clientId.String()))
+						continue
+					}
+
 					clientCtx := utils.ContextFromIdentifiers(mainCtx, clientId)
 					err := client.SendReconciliationRequestMessage(clientCtx)
 					if err != nil {

utils/utils.go

@@ -7,14 +7,16 @@ import (
 	"errors"
 	"flag"
 	"fmt"
-	"github.com/cenkalti/backoff/v4"
 	"net/http"
 	"net/http/pprof"
 	"path/filepath"
 	"reflect"
 	"strconv"
 	"time"
 
+	"github.com/cenkalti/backoff/v4"
+	"golang.org/x/mod/semver"
+
 	"github.com/SergJa/jsonhash"
 	"github.com/apache/pulsar-client-go/pulsar"
 	"github.com/davecgh/go-spew/spew"
@@ -272,3 +274,18 @@ func NewBackOff() backoff.BackOff {
 	b.MaxElapsedTime = 0
 	return b
 }
+
+// GreaterOrEqualVersion returns true if a version is greater or equal to b
+func GreaterOrEqualVersion(a string, b string) bool {
+	return semver.Compare(a, b) >= 0
+}
+
+func IsBatchMessageSupported(version string) bool {
+	const minimumSupportedVersion = "v0.0.57"
+
+	if version == "" {
+		return false
+	}
+
+	return GreaterOrEqualVersion(version, minimumSupportedVersion)
+}

utils/utils_test.go

@@ -134,3 +134,44 @@ func TestRemoveManagedFields(t *testing.T) {
 		})
 	}
 }
+
+func TestGreaterOrEqualVersion(t *testing.T) {
+	testCases := []struct {
+		a        string
+		b        string
+		expected bool
+	}{
+		{"v0.0.2", "v0.0.1", true},
+		{"v0.0.1", "v0.0.2", false},
+		{"v0.0.1", "v0.0.1", true},
+	}
+
+	for _, tc := range testCases {
+		result := GreaterOrEqualVersion(tc.a, tc.b)
+		if result != tc.expected {
+			t.Errorf("For version %s >= %s, expected %v but got %v", tc.a, tc.b, tc.expected, result)
+		}
+	}
+}
+
+func TestIsBatchMessageSupported(t *testing.T) {
+	testCases := []struct {
+		version  string
+		expected bool
+	}{
+		{"", false},                // Empty version should return false
+		{"v0.0.56", false},         // Version less than the minimum supported version should return false
+		{"v0.0.57", true},          // Minimum supported version should return true
+		{"v0.0.58", true},          // Version greater than the minimum supported version should return true
+		{"v1.0.0", true},           // Version with a major version greater than 0 should return true
+		{"v1.2.3", true},           // Version with a major version greater than 0 should return true
+		{"invalid_version", false}, // Invalid version should return false
+	}
+
+	for _, tc := range testCases {
+		result := IsBatchMessageSupported(tc.version)
+		if result != tc.expected {
+			t.Errorf("For version %s, expected %v but got %v", tc.version, tc.expected, result)
+		}
+	}
+}