Fix kafka_franz input batching period

CHANGELOG.md

@@ -5,6 +5,8 @@ All notable changes to this project will be documented in this file.
 
 ## Unreleased
 
+## 4.22.0 - 2023-10-03
+
 ### Added
 
 - The `-e/--env-file` cli flag for importing environment variable files now supports glob patterns.
@@ -23,6 +25,7 @@ All notable changes to this project will be documented in this file.
 - The `sqs` input now periodically refreshes the visibility timeout of messages that take a significant amount of time to process.
 - The `ts_add_iso8601` and `ts_sub_iso8601` bloblang methods now return the correct error for certain invalid durations.
 - The `discord` output no longer ignores structured message fields containing underscores.
+- Fixed an issue where the `kafka_franz` input was ignoring batching periods and stalling.
 
 ### Changed
 

internal/batch/policy/policy.go

@@ -188,7 +188,11 @@ func (p *Batcher) UntilNext() time.Duration {
 	if p.period <= 0 {
 		return -1
 	}
-	return time.Until(p.lastBatch.Add(p.period))
+	tUntil := time.Until(p.lastBatch.Add(p.period))
+	if tUntil < 0 {
+		tUntil = 0
+	}
+	return tUntil
 }
 
 //------------------------------------------------------------------------------

internal/impl/kafka/input_kafka_franz.go

@@ -376,7 +376,7 @@ func (p *partitionTracker) loop() {
 					return
 				}
 
-				if sendBatch, _ := p.batcher.Flush(closeAtLeisureCtx); len(sendBatch) == 0 {
+				if sendBatch, _ = p.batcher.Flush(closeAtLeisureCtx); len(sendBatch) == 0 {
 					return
 				}
 				sendRecord = p.topBatchRecord