[fix][fn]Reset idle timer correctly (apache#20450)

Co-authored-by: Andy Walker <andy@andy.dev>

Fix apache#20449

<!-- or this PR is one task of an issue -->

Master Issue: apache#20449

### Verifying this change

- [ ] Make sure that the change passes the CI checks.

This change is a trivial rework / code cleanup without any test coverage.

### Does this pull request potentially affect one of the following parts:

<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->

*If the box was checked, please highlight the changes*

- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment

### Documentation

<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->

- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [X] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->

### Matching PR in forked repository

PR in forked repository: flowchartsman#5

pulsar-function-go/pf/instance.go

@@ -149,7 +149,6 @@ func (gi *goInstance) startFunction(function function) error {
 	defer metricsServicer.close()
 CLOSE:
 	for {
-		idleTimer.Reset(idleDuration)
 		select {
 		case cm := <-channel:
 			msgInput := cm.Message
@@ -181,6 +180,11 @@ CLOSE:
 			close(channel)
 			break CLOSE
 		}
+		// reset the idle timer and drain if appropriate before the next loop
+		if !idleTimer.Stop() {
+			<-idleTimer.C
+		}
+		idleTimer.Reset(idleDuration)
 	}
 
 	gi.closeLogTopic()