chore: resolved-lint-error

Author: saksham-datazip

drivers/kafka/internal/cdc.go

@@ -91,7 +91,7 @@ func (k *Kafka) StreamChanges(ctx context.Context, readerID int, metadataStates
 
 	err = k.processKafkaMessages(ctx, reader, func(record types.KafkaRecord) (bool, error) {
 		// get current partition metadata and key
-		currentPartitionKey := types.PartitionKey{Topic: record.Message.Topic, Partition: int(record.Message.Partition)}
+		currentPartitionKey := types.PartitionKey{Topic: record.Message.Topic, Partition: record.Message.Partition}
 		currentPartitionMeta, exists := k.readerManager.GetPartitionIndex(fmt.Sprintf("%s:%d", record.Message.Topic, record.Message.Partition))
 		if !exists {
 			return false, fmt.Errorf("missing partition index for topic %s partition %d", record.Message.Topic, record.Message.Partition)
@@ -197,19 +197,19 @@ func (k *Kafka) PostCDC(ctx context.Context, readerIdx int) error {
 // for processing messages from a Kafka reader.
 func (k *Kafka) processKafkaMessages(ctx context.Context, reader *kgo.Client, stopProcessFn func(record types.KafkaRecord) (bool, error)) error {
 	for {
+		// Rebalance/exit checks must run even when PollFetches returns an empty batch;
+		// otherwise a reader reassigned to empty partitions never enters the record loop.
+		if stopProcessing, err := k.readerManager.FetchExitState(); stopProcessing {
+			return err
+		}
+
 		pollCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
 		messages := reader.PollFetches(pollCtx)
 		pollCtxErr := pollCtx.Err()
 		cancel()
 
 		pollFetchErrors := messages.Errors()
 
-		// Rebalance/exit checks must run even when PollFetches returns an empty batch;
-		// otherwise a reader reassigned to empty partitions never enters the record loop.
-		if stopProcessing, err := k.readerManager.FetchExitState(); stopProcessing {
-			return err
-		}
-
 		// return early if poll context is deadline exceeded
 		if errors.Is(pollCtxErr, context.DeadlineExceeded) {
 			logger.Warnf("poll context deadline exceeded: %v", pollCtxErr)

drivers/kafka/internal/kafka.go

@@ -425,7 +425,7 @@ func (k *Kafka) getReaderAssignedPartitions(ctx context.Context, readerIndex int
 
 			for _, topic := range assignment.Topics {
 				for _, partition := range topic.Partitions {
-					assigned = append(assigned, types.PartitionKey{Topic: topic.Topic, Partition: int(partition)})
+					assigned = append(assigned, types.PartitionKey{Topic: topic.Topic, Partition: partition})
 				}
 			}
 		}

pkg/kafka/balancer.go

@@ -59,7 +59,7 @@ func (b *CustomGroupBalancer) Balance(consumerBalancer *kgo.ConsumerBalancer, pa
 	for topic, partitionCount := range partitionsPerTopic {
 		for partition := int32(0); partition < partitionCount; partition++ {
 			if _, ok := b.partitionIndex[fmt.Sprintf("%s:%d", topic, partition)]; ok {
-				activePartitions = append(activePartitions, types.PartitionKey{Topic: topic, Partition: int(partition)})
+				activePartitions = append(activePartitions, types.PartitionKey{Topic: topic, Partition: partition})
 			}
 		}
 	}
@@ -70,7 +70,7 @@ func (b *CustomGroupBalancer) Balance(consumerBalancer *kgo.ConsumerBalancer, pa
 	})
 
 	for currentIndex, activePartition := range activePartitions {
-		plan.AddPartition(&members[currentIndex%consumerCount], activePartition.Topic, int32(activePartition.Partition))
+		plan.AddPartition(&members[currentIndex%consumerCount], activePartition.Topic, activePartition.Partition)
 	}
 	return plan
 }

pkg/kafka/reader.go

@@ -61,12 +61,12 @@ func (r *ReaderManager) CreateReaders(ctx context.Context, streams []types.Strea
 	return r.waitForConsumerGroupJoin(consumerGroupID)
 }
 
-// GetReaders returns the created readers
+// GetReader returns the created readers
 func (r *ReaderManager) GetReader(readerID int) *kgo.Client {
 	return r.readers[readerID].reader
 }
 
-// GetReaders returns the created readers
+// GetReaderCount returns the created readers count
 func (r *ReaderManager) GetReaderCount() int {
 	return len(r.readers)
 }
@@ -125,7 +125,7 @@ func (r *ReaderManager) SetPartitions(ctx context.Context, stream types.StreamIn
 			continue
 		}
 
-		committedOffset, hasCommittedOffset := committedTopicOffsets[int(partition.Partition)]
+		committedOffset, hasCommittedOffset := committedTopicOffsets[partition.Partition]
 
 		// check if the partition has any messages at all, if not then skip
 		if startOffset.Offset >= endOffset.Offset {
@@ -141,7 +141,7 @@ func (r *ReaderManager) SetPartitions(ctx context.Context, stream types.StreamIn
 
 		r.partitionIndex[fmt.Sprintf("%s:%d", topic, partition.Partition)] = types.PartitionMetaData{
 			Stream:      stream,
-			PartitionID: int(partition.Partition),
+			PartitionID: partition.Partition,
 			EndOffset:   endOffset.Offset,
 		}
 	}
@@ -163,19 +163,19 @@ func (r *ReaderManager) GetTopicMetadata(ctx context.Context, topic string) (*ka
 }
 
 // FetchCommittedOffsets fetches committed offsets for a topic
-func (r *ReaderManager) FetchCommittedOffsets(ctx context.Context, topic string, partitions map[int32]kadm.PartitionDetail) (map[int]int64, error) {
+func (r *ReaderManager) FetchCommittedOffsets(ctx context.Context, topic string, partitions map[int32]kadm.PartitionDetail) (map[int32]int64, error) {
 	offsets, err := r.config.AdminClient.FetchOffsets(ctx, r.config.ConsumerGroupID)
 	if err != nil {
 		return nil, fmt.Errorf("could not fetch committed offsets for group %s", r.config.ConsumerGroupID)
 	}
 
-	committedTopicOffsets := make(map[int]int64)
+	committedTopicOffsets := make(map[int32]int64)
 	for _, partitionDetail := range partitions {
 		offset, exists := offsets.Lookup(topic, partitionDetail.Partition)
 		if !exists {
 			continue
 		}
-		committedTopicOffsets[int(partitionDetail.Partition)] = offset.At
+		committedTopicOffsets[partitionDetail.Partition] = offset.At
 	}
 	return committedTopicOffsets, nil
 }
@@ -334,11 +334,6 @@ func (r *ReaderManager) CreateReader(streams []types.StreamInterface, consumerGr
 	return reader, nil
 }
 
-// GenerationID returns the consumer-group generation stored after CreateReaders join wait.
-func (r *ReaderManager) GenerationID() int32 {
-	return r.generationID.Load()
-}
-
 // RebalanceDetected is true when the client's group generation differs from the stored baseline.
 func (r *ReaderManager) RebalanceDetected(client *kgo.Client) bool {
 	_, generationID := client.GroupMetadata()
@@ -383,7 +378,6 @@ func (r *ReaderManager) waitForConsumerGroupJoin(consumerGroupID string) error {
 			} else if expectedGenerationID < 0 {
 				expectedGenerationID = generationID
 			}
-
 		}
 
 		if allReadersReady && expectedGenerationID >= 0 {

types/kafka_types.go

@@ -17,14 +17,14 @@ const (
 type PartitionMetaData struct {
 	ReaderID    string
 	Stream      StreamInterface
-	PartitionID int
+	PartitionID int32
 	EndOffset   int64
 }
 
 // PartitionKey represents a unique key for a Kafka partition and topic
 type PartitionKey struct {
 	Topic     string
-	Partition int
+	Partition int32
 }
 
 // KafkaRecord represents a record (data + message) from a Kafka partition