Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

All flush failures are retryable except for timestamp formatting. Fix… #287

Open
wants to merge 1 commit into
base: mainline
Choose a base branch
from
Open

All flush failures are retryable except for timestamp formatting. Fix… #287

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 2 additions & 3 deletions kinesis/kinesis.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -291,14 +291,13 @@ func (outputPlugin *OutputPlugin) AddRecord(records *[]*kinesis.PutRecordsReques
return fluentbit.FLB_OK
}

// FlushAggregatedRecords must be called after
// Returns FLB_OK, FLB_RETRY, FLB_ERROR
func (outputPlugin *OutputPlugin) FlushAggregatedRecords(records *[]*kinesis.PutRecordsRequestEntry) int {

aggRecord, err := outputPlugin.aggregator.AggregateRecords()
if err != nil {
logrus.Errorf("[kinesis %d] Failed to aggregate record %v\n", outputPlugin.PluginID, err)
return fluentbit.FLB_ERROR
return fluentbit.FLB_RETRY
}
Comment on lines 295 to 301
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm. Not sure about this one... If aggregation fails once, won't it probably fail again the next time?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point, we could spend more time investigating this, I don't have time right now with higher priority issues.

If there is any way for it to fail and not fail again then we should always return FLB_RETRY so I think if we don't know then we should return retry.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No comment on it in original PR :/ #60


if aggRecord != nil {
Expand Down Expand Up @@ -384,7 +383,7 @@ func (outputPlugin *OutputPlugin) FlushWithRetries(count int, records []*kinesis

switch retCode {
case output.FLB_ERROR:
logrus.Errorf("[kinesis %d] Failed to send (%d) records with error", outputPlugin.PluginID, len(records))
logrus.Errorf("[kinesis %d] Failed to send (%d) records, unretryable error", outputPlugin.PluginID, len(records))
case output.FLB_RETRY:
logrus.Errorf("[kinesis %d] Failed to send (%d) records after retries %d", outputPlugin.PluginID, len(records), outputPlugin.concurrencyRetryLimit)
case output.FLB_OK:
Expand Down