func (s *MapperSuite) TestChannelFull(c *C) {
	clock.Freeze(testEpoch)
	r := &mockResolver{
		w2b: map[Worker]*sarama.Broker{
			s.workers[0]: s.brokers[4],
		},
	}
	s.workers[0].assignmentsCh = make(chan Executor, 1)
	s.workers[0].assignmentsCh <- &mockExecutor{id: 42}
	s.cfg.Consumer.RetryBackoff = 500 * clock.Millisecond
	m := Spawn(s.ns, s.cfg, r)
	m.OnWorkerSpawned(s.workers[0])
	c.Assert(clock.Wait4Scheduled(1, 3*clock.Second), Equals, true)
	clock.Advance(1999 * clock.Millisecond)
	<-s.workers[0].assignmentsCh
	clock.Advance(1 * clock.Millisecond)
	assigned := <-s.workers[0].assignmentsCh
	c.Assert(assigned.BrokerConn(), Equals, s.brokers[4])
	m.OnWorkerStopped(s.workers[0])
	m.Stop()
	assertNoAssignments(c, s.workers[0], 0)
}
If writing to the workers assignments channel would block, then the
resolution attempt is considered to fail, and retry is scheduled.
package mapper
import (
	"fmt"
	"sync/atomic"
	"testing"
	"time"
	"github.com/Shopify/sarama"
	"github.com/mailgun/holster/v4/clock"
	"github.com/mailgun/kafka-pixy/actor"
	"github.com/mailgun/kafka-pixy/config"
	"github.com/mailgun/kafka-pixy/none"
	"github.com/mailgun/kafka-pixy/testhelpers"
	"github.com/pkg/errors"
	log "github.com/sirupsen/logrus"
	. "gopkg.in/check.v1"
)
var testEpoch = clock.Date(2009, 02, 19, 13, 45, 0, 0, clock.UTC)
func Test(t *testing.T) {
	TestingT(t)
}
type MapperSuite struct {
	ns      *actor.Descriptor
	cfg     *config.Proxy
	workers []*mockWorker
	brokers []*sarama.Broker
}
var (
	_          = Suite(&MapperSuite{})
	lastExecId int32
)
func (s *MapperSuite) SetUpSuite(c *C) {
	testMode = true
	testhelpers.InitLogging()
}
func (s *MapperSuite) TearDownSuite(c *C) {
	testMode = false
}
func (s *MapperSuite) SetUpTest(c *C) {
	s.ns = actor.Root().NewChild("T")
	s.cfg = testhelpers.NewTestProxyCfg("mapper")
	s.workers = make([]*mockWorker, 5)
	s.brokers = make([]*sarama.Broker, 5)
	for i := 0; i < 5; i++ {
		s.workers[i] = &mockWorker{i, make(chan Executor, 10)}
		s.brokers[i] = sarama.NewBroker(fmt.Sprintf("b%d", i))
	}
}
func (s *MapperSuite) TestBasicAssignment(c *C) {
	r := &mockResolver{
		w2b: map[Worker]*sarama.Broker{
			s.workers[0]: s.brokers[1],
			s.workers[1]: s.brokers[0],
			s.workers[2]: s.brokers[0],
			s.workers[3]: s.brokers[1],
			s.workers[4]: s.brokers[0],
		},
	}
	m := Spawn(s.ns, s.cfg, r)
	for _, w := range s.workers {
		m.OnWorkerSpawned(w)
	}
	assigned := make([]Executor, len(s.workers))
	for i, w := range s.workers {
		assigned[i] = <-w.assignmentsCh
	}
	c.Assert(assigned[0].BrokerConn(), Equals, s.brokers[1])
	c.Assert(assigned[1].BrokerConn(), Equals, s.brokers[0])
	c.Assert(assigned[3], Equals, assigned[0])
	c.Assert(assigned[2], Equals, assigned[1])
	c.Assert(assigned[4], Equals, assigned[1])
	for _, w := range s.workers {
		m.OnWorkerStopped(w)
	}
	m.Stop()
	for _, w := range s.workers {
		assertNoAssignments(c, w, 0)
	}
	assertExecutorStopped(c, assigned[0], 200*clock.Millisecond)
	assertExecutorStopped(c, assigned[1], 200*clock.Millisecond)
}
func (s *MapperSuite) TestTriggerReassignImmediate(c *C) {
	clock.Freeze(testEpoch)
	r := &mockResolver{
		w2b: map[Worker]*sarama.Broker{
			s.workers[0]: s.brokers[0],
			s.workers[1]: s.brokers[1],
		},
	}
	s.cfg.Consumer.RetryBackoff = 500 * clock.Millisecond
	m := Spawn(s.ns, s.cfg, r)
	assigned := make([]Executor, len(s.workers))
	m.OnWorkerSpawned(s.workers[0])
	m.OnWorkerSpawned(s.workers[1])
	assigned[0] = <-s.workers[0].assignmentsCh
	assigned[1] = <-s.workers[1].assignmentsCh
	clock.Advance(500 * clock.Millisecond)
	log.Info("*** When")
	r.w2b = map[Worker]*sarama.Broker{
		s.workers[0]: s.brokers[1],
		s.workers[1]: s.brokers[1],
	}
	skipEvents(1)
	m.TriggerReassign(s.workers[0])
	m.TriggerReassign(s.workers[1])  
	log.Info("*** Then")
	c.Assert((<-s.workers[0].assignmentsCh).BrokerConn(), Equals, s.brokers[1])
	c.Assert((<-s.workers[1].assignmentsCh).BrokerConn(), Equals, s.brokers[1])
	assertExecutorStopped(c, assigned[0], 200*clock.Millisecond)
	m.OnWorkerStopped(s.workers[0])
	m.OnWorkerStopped(s.workers[1])
	m.Stop()
	assertNoAssignments(c, s.workers[0], 0)
	assertNoAssignments(c, s.workers[1], 0)
}
func (s *MapperSuite) TestTriggerReassignDeferred(c *C) {
	clock.Freeze(testEpoch)
	r := &mockResolver{
		w2b: map[Worker]*sarama.Broker{
			s.workers[0]: s.brokers[1],
			s.workers[1]: s.brokers[0],
			s.workers[2]: s.brokers[0],
			s.workers[3]: s.brokers[1],
			s.workers[4]: s.brokers[0],
		},
	}
	s.cfg.Consumer.RetryBackoff = 500 * clock.Millisecond
	m := Spawn(s.ns, s.cfg, r)
	assigned := make([]Executor, len(s.workers))
	for i, w := range s.workers {
		m.OnWorkerSpawned(w)
		assigned[i] = <-w.assignmentsCh
	}
	log.Info("*** When")
	r.w2b = map[Worker]*sarama.Broker{
		s.workers[0]: s.brokers[2],
		s.workers[1]: s.brokers[0],
		s.workers[2]: s.brokers[0],
		s.workers[3]: s.brokers[0],
		s.workers[4]: s.brokers[0],
	}
	for _, w := range s.workers {
		m.TriggerReassign(w)
	}
	m.TriggerReassign(s.workers[0])
	m.TriggerReassign(s.workers[1])
	assertNoAssignments(c, s.workers[1], 100*clock.Millisecond)
	log.Info("*** Then")
	clock.Advance(499 * clock.Millisecond)
	for _, w := range s.workers {
		assertNoAssignments(c, w, 0)
	}
	clock.Advance(1 * clock.Millisecond)
	c.Assert((<-s.workers[0].assignmentsCh).BrokerConn(), Equals, s.brokers[2])
	c.Assert((<-s.workers[1].assignmentsCh).BrokerConn(), Equals, s.brokers[0])
	c.Assert((<-s.workers[2].assignmentsCh).BrokerConn(), Equals, s.brokers[0])
	c.Assert((<-s.workers[3].assignmentsCh).BrokerConn(), Equals, s.brokers[0])
	c.Assert((<-s.workers[4].assignmentsCh).BrokerConn(), Equals, s.brokers[0])
	c.Assert(assigned[0].BrokerConn(), Equals, s.brokers[1])
	assertExecutorStopped(c, assigned[0], 200*clock.Millisecond)
	for _, w := range s.workers {
		m.OnWorkerStopped(w)
	}
	m.Stop()
	for _, w := range s.workers {
		assertNoAssignments(c, w, 0)
	}
}
func (s *MapperSuite) TestOnlyOneReassign(c *C) {
	clock.Freeze(testEpoch)
	r := &mockResolver{
		w2b: map[Worker]*sarama.Broker{
			s.workers[0]: s.brokers[2],
			s.workers[1]: s.brokers[2],
		},
	}
	s.cfg.Consumer.RetryBackoff = 500 * clock.Millisecond
	m := Spawn(s.ns, s.cfg, r)
	m.OnWorkerSpawned(s.workers[0])
	m.OnWorkerSpawned(s.workers[1])
	c.Assert((<-s.workers[0].assignmentsCh).BrokerConn(), Equals, s.brokers[2])
	c.Assert((<-s.workers[1].assignmentsCh).BrokerConn(), Equals, s.brokers[2])
	log.Info("*** When")
	m.TriggerReassign(s.workers[0])
	c.Assert(clock.Wait4Scheduled(1, 3*clock.Second), Equals, true)
	clock.Advance(200 * clock.Millisecond)
	m.TriggerReassign(s.workers[0])
	m.TriggerReassign(s.workers[1])
	assertNoAssignments(c, s.workers[1], 100*clock.Millisecond)
	clock.Advance(100 * clock.Millisecond)
	m.TriggerReassign(s.workers[0])
	m.TriggerReassign(s.workers[1])
	clock.Advance(199 * clock.Millisecond)
	m.TriggerReassign(s.workers[1])
	c.Assert(clock.Wait4Scheduled(1, 0), Equals, true)
	c.Assert(clock.Wait4Scheduled(2, 100*clock.Millisecond), Equals, false)
	assertNoAssignments(c, s.workers[0], 0)
	assertNoAssignments(c, s.workers[1], 0)
	log.Info("*** Then")
	clock.Advance(1 * clock.Millisecond)
	c.Assert((<-s.workers[0].assignmentsCh).BrokerConn(), Equals, s.brokers[2])
	assertNoAssignments(c, s.workers[1], 0)
	clock.Advance(199 * clock.Millisecond)
	assertNoAssignments(c, s.workers[1], 0)
	clock.Advance(1 * clock.Millisecond)
	c.Assert((<-s.workers[1].assignmentsCh).BrokerConn(), Equals, s.brokers[2])
	clock.Advance(1000 * clock.Millisecond)
	assertNoAssignments(c, s.workers[0], 0)
	assertNoAssignments(c, s.workers[1], 0)
	log.Info("*** Cleanup")
	m.OnWorkerStopped(s.workers[0])
	m.OnWorkerStopped(s.workers[1])
	m.Stop()
	assertNoAssignments(c, s.workers[0], 0)
	assertNoAssignments(c, s.workers[1], 0)
}
func (s *MapperSuite) TestDuplicateSpawn(c *C) {
	r := &mockResolver{
		w2b: map[Worker]*sarama.Broker{
			s.workers[0]: s.brokers[0],
		},
	}
	m := Spawn(s.ns, s.cfg, r)
	m.OnWorkerSpawned(s.workers[0])
	assigned := <-s.workers[0].assignmentsCh
	c.Assert(assigned.BrokerConn(), Equals, s.brokers[0])
	m.OnWorkerSpawned(s.workers[0])
	m.OnWorkerSpawned(s.workers[0])
	m.OnWorkerSpawned(s.workers[0])
	assertNoAssignments(c, s.workers[0], 0)
	m.OnWorkerStopped(s.workers[0])
	m.Stop()
}
func (s *MapperSuite) TestNeverAssigned(c *C) {
	r := &mockResolver{
		errors: map[Worker]error{
			s.workers[0]: errors.New("Kaboom!"),
		},
	}
	s.cfg.Consumer.RetryBackoff = 50 * clock.Millisecond
	m := Spawn(s.ns, s.cfg, r)
	m.OnWorkerSpawned(s.workers[0])
	assertNoAssignments(c, s.workers[0], 200*clock.Millisecond)
	m.OnWorkerStopped(s.workers[0])
	m.Stop()
	assertNoAssignments(c, s.workers[0], 0)
}
func (s *MapperSuite) TestNilResolved(c *C) {
	clock.Freeze(testEpoch)
	r := &mockResolver{
		w2b: map[Worker]*sarama.Broker{
			s.workers[0]: nil,
		},
	}
	s.cfg.Consumer.RetryBackoff = 500 * clock.Millisecond
	m := Spawn(s.ns, s.cfg, r)
	m.OnWorkerSpawned(s.workers[0])
	c.Assert(clock.Wait4Scheduled(1, 3*clock.Second), Equals, true)
	clock.Advance(499 * clock.Millisecond)
	assertNoAssignments(c, s.workers[0], 0)
	r.w2b = map[Worker]*sarama.Broker{
		s.workers[0]: s.brokers[3],
	}
	clock.Advance(1 * clock.Millisecond)
	assigned := <-s.workers[0].assignmentsCh
	c.Assert(assigned.BrokerConn(), Equals, s.brokers[3])
	m.OnWorkerStopped(s.workers[0])
	m.Stop()
	assertNoAssignments(c, s.workers[0], 0)
}