refactor kafka reader && expose error for kafka and mongo
package kafka
import (
- "encoding/json"
"os"
"sync"
"testing"
@@ -31,9 +30,7 @@ func TestKafkaReader(t *testing.T) {
ZookeeperPeers: []string{"localhost:2181"},
ZookeeperTimeout: time.Second,
Whence: "oldest",
- readChan: make(chan json.RawMessage),
errs: make(chan error, 1000),
- status: reader.StatusInit,
mux: new(sync.Mutex),
started: false,
statsLock: new(sync.RWMutex),
