Error on uninitialize brokers

Author: jakeloo

internal/storage/kafka_postgres.go

@@ -55,13 +55,9 @@ func NewKafkaPostgresConnector(cfg *config.KafkaConfig) (*KafkaPostgresConnector
 	}
 
 	// Initialize Kafka publisher if enabled
-	var kafkaPublisher *KafkaPublisher
-	if cfg.Brokers != "" {
-		kafkaPublisher, err = NewKafkaPublisher(cfg)
-		if err != nil {
-			log.Warn().Err(err).Msg("Failed to initialize Kafka publisher, continuing without publishing")
-			kafkaPublisher = nil
-		}
+	kafkaPublisher, err := NewKafkaPublisher(cfg)
+	if err != nil {
+		return nil, err
 	}
 
 	return &KafkaPostgresConnector{
@@ -502,11 +498,9 @@ func (kp *KafkaPostgresConnector) ReplaceBlockData(data []common.BlockData) ([]c
 	oldBlocks := []common.BlockData{}
 
 	// Publish reorg event to Kafka
-	if kp.kafkaPublisher != nil {
-		// Publish new blocks (the reorg handler will mark old ones as reverted)
-		if err := kp.kafkaPublisher.PublishBlockData(data); err != nil {
-			return nil, fmt.Errorf("failed to publish reorg blocks to kafka: %w", err)
-		}
+	// TODO: Publish new blocks (the reorg handler will mark old ones as reverted)
+	if err := kp.kafkaPublisher.PublishBlockData(data); err != nil {
+		return nil, fmt.Errorf("failed to publish reorg blocks to kafka: %w", err)
 	}
 
 	// Update cursor to track the highest block number

internal/storage/kafka_publisher.go

@@ -126,7 +126,7 @@ func (p *KafkaPublisher) publishMessages(ctx context.Context, messages []*kgo.Re
 }
 
 func (p *KafkaPublisher) publishBlockData(blockData []common.BlockData, isReorg bool) error {
-	if p.client == nil || len(blockData) == 0 {
+	if len(blockData) == 0 {
 		return nil
 	}
 
@@ -135,6 +135,7 @@ func (p *KafkaPublisher) publishBlockData(blockData []common.BlockData, isReorg
 	// Prepare messages for blocks, events, transactions and traces
 	blockMessages := make([]*kgo.Record, len(blockData))
 
+	// TODO: handle reorg
 	status := "new"
 	if isReorg {
 		status = "reverted"