mirror of
https://github.com/hanzoai/traces.git
synced 2026-10-30 18:20:41 +00:00
edeb184ac2
* feat: handle kafka errors that require to refresh metadata * log the error * retry on list offsets errors * add unkown broker error * make it retryable * force metadata refresh on unknown broker error * force metadata refres * fix test * simplify retry logic * Update modules/blockbuilder/blockbuilder.go Co-authored-by: Mario <[email protected]> * fix typo * fix nil exception * handle metadata refresh error when commiting offsets --------- Co-authored-by: Mario <[email protected]>
40 lines
932 B
Go
40 lines
932 B
Go
package ingest
|
|
|
|
import (
|
|
"errors"
|
|
"testing"
|
|
|
|
"github.com/stretchr/testify/require"
|
|
"github.com/twmb/franz-go/pkg/kerr"
|
|
)
|
|
|
|
func TestHandleKafkaError(t *testing.T) {
|
|
tests := []struct {
|
|
err error
|
|
expectedRefresh bool
|
|
}{
|
|
{nil, false},
|
|
{errors.New("Some error"), false},
|
|
{errors.New("unknown broker"), true},
|
|
{kerr.NotLeaderForPartition, true},
|
|
{kerr.ReplicaNotAvailable, true},
|
|
{kerr.UnknownLeaderEpoch, true},
|
|
{kerr.LeaderNotAvailable, true},
|
|
{kerr.BrokerNotAvailable, true},
|
|
{kerr.UnknownTopicOrPartition, true},
|
|
{kerr.NetworkException, true},
|
|
{kerr.NotCoordinator, true},
|
|
{kerr.IllegalSaslState, false},
|
|
}
|
|
|
|
for _, test := range tests {
|
|
refreshCalled := false
|
|
refreshFunc := func() {
|
|
refreshCalled = true
|
|
}
|
|
|
|
HandleKafkaError(test.err, refreshFunc)
|
|
require.Equal(t, test.expectedRefresh, refreshCalled, "HandleKafkaError(%v) refresh function call mismatch", test.err)
|
|
}
|
|
}
|