chore: small fixes for #16348 (#16358)

Author: fcjack

pkg/dataobj/consumer/config.go

@@ -30,9 +30,5 @@ func (cfg *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
 	cfg.BuilderConfig.RegisterFlagsWithPrefix(prefix, f)
 	cfg.UploaderConfig.RegisterFlagsWithPrefix(prefix, f)
 
-	if cfg.IdleFlushTimeout <= 0 {
-		cfg.IdleFlushTimeout = 60 * 60 * time.Second // default to 1 hour
-	}
-
-	f.DurationVar(&cfg.IdleFlushTimeout, prefix+"idle-flush-timeout", cfg.IdleFlushTimeout, "The maximum amount of time to wait in seconds before flushing an object that is no longer receiving new writes")
+	f.DurationVar(&cfg.IdleFlushTimeout, prefix+"idle-flush-timeout", 60*60*time.Second, "The maximum amount of time to wait in seconds before flushing an object that is no longer receiving new writes")
 }

pkg/dataobj/consumer/partition_processor.go

@@ -40,8 +40,8 @@ type partitionProcessor struct {
 	bufPool     *sync.Pool
 
 	// Idle stream handling
-	idleFlushTimout time.Duration
-	lastFlush       time.Time
+	idleFlushTimeout time.Duration
+	lastFlush        time.Time
 
 	// Metrics
 	metrics *partitionOffsetMetrics
@@ -113,7 +113,7 @@ func newPartitionProcessor(
 		uploader:         uploader,
 		metastoreManager: metastoreManager,
 		bufPool:          bufPool,
-		idleFlushTimout:  idleFlushTimeout,
+		idleFlushTimeout: idleFlushTimeout,
 		lastFlush:        time.Now(),
 	}
 }
@@ -136,7 +136,7 @@ func (p *partitionProcessor) start() {
 				}
 				p.processRecord(record)
 
-			case <-time.After(p.idleFlushTimout):
+			case <-time.After(p.idleFlushTimeout):
 				p.idleFlush()
 			}
 		}
@@ -294,8 +294,7 @@ func (p *partitionProcessor) idleFlush() {
 		return
 	}
 
-	now := time.Now()
-	if now.Sub(p.lastFlush) < p.idleFlushTimout {
+	if time.Since(p.lastFlush) < p.idleFlushTimeout {
 		return // Avoid checking too frequently
 	}
 
@@ -310,6 +309,6 @@ func (p *partitionProcessor) idleFlush() {
 			return
 		}
 
-		p.lastFlush = now
+		p.lastFlush = time.Now()
 	}()
 }