public void CancellationShouldInterruptConsumption()
        {
            var routerProxy = new BrokerRouterProxy(new MoqMockingKernel());
            routerProxy.BrokerConn0.FetchResponseFunction = () => { return new FetchResponse(); };

            var router = routerProxy.Create();

            var options = CreateOptions(router);

            using (var consumer = new Consumer(options))
            {
                var tokenSrc = new CancellationTokenSource();

                var consumeTask = Task.Run(() => consumer.Consume(tokenSrc.Token).FirstOrDefault());

                //wait until the fake broker is running and requesting fetches
                TaskTest.WaitFor(() => routerProxy.BrokerConn0.FetchRequestCallCount > 10);

                tokenSrc.Cancel();

                Assert.That(
                    Assert.Throws<AggregateException>(consumeTask.Wait).InnerException,
                    Is.TypeOf<OperationCanceledException>());
            }
        }

        public async Task ShouldTryToRefreshMataDataIfCanRecoverByRefreshMetadata(ErrorResponseCode code)
        {
            var routerProxy = new BrokerRouterProxy(_kernel);
            routerProxy._cacheExpiration = new TimeSpan(10);
            var router = routerProxy.Create();

            int partitionId = 0;

            ProtocolGateway protocolGateway = new ProtocolGateway(router);
            var fetchRequest = new FetchRequest();
            bool sendExOnFirstTime = true;

            Func<Task<FetchResponse>> ShouldReturnErrorAndThenNoError = async () =>
            {
                Task.Delay(routerProxy._cacheExpiration).Wait();
                Task.Delay(1).Wait();
                if (sendExOnFirstTime)
                {
                    sendExOnFirstTime = false;
                    return new FetchResponse() { Error = (short)code };
                }
                return new FetchResponse() { Error = (short)ErrorResponseCode.NoError };
            };
            routerProxy.BrokerConn0.FetchResponseFunction = ShouldReturnErrorAndThenNoError;
            routerProxy.BrokerConn0.MetadataResponseFunction = BrokerRouterProxy.DefaultMetadataResponse;

            await protocolGateway.SendProtocolRequest(fetchRequest, BrokerRouterProxy.TestTopic, partitionId);

            Assert.That(routerProxy.BrokerConn0.MetadataRequestCallCount, Is.EqualTo(2));
            Assert.That(routerProxy.BrokerConn0.FetchRequestCallCount, Is.EqualTo(2));
        }

        public void EmptyTopicMetadataShouldThrowException()
        {
            var routerProxy = new BrokerRouterProxy(_kernel);
            var router = routerProxy.Create();
            var common = new CommonQueries(router);

            common.GetTopic("MissingTopic");
        }

 public async Task ShouldThrowFormatExceptionWhenTopicIsInvalid()
 {
     var routerProxy = new BrokerRouterProxy(_kernel);
     var router = routerProxy.Create();
     string invalidTopic = " ";
     var fetchRequest = new FetchRequest();
     ProtocolGateway protocolGateway = new ProtocolGateway(router);
     await protocolGateway.SendProtocolRequest(fetchRequest, invalidTopic, 0);
 }

        public void GetTopicShouldReturnTopic()
        {
            var routerProxy = new BrokerRouterProxy(_kernel);
            var router = routerProxy.Create();
            var common = new CommonQueries(router);

            var result = common.GetTopic(BrokerRouterProxy.TestTopic);
            Assert.That(result.Name, Is.EqualTo(BrokerRouterProxy.TestTopic));
        }

        public void GetTopicOffsetShouldQueryEachBroker()
        {
            var routerProxy = new BrokerRouterProxy(_kernel);
            var router = routerProxy.Create();
            var common = new CommonQueries(router);

            var result = common.GetTopicOffsetAsync(BrokerRouterProxy.TestTopic).Result;
            Assert.That(routerProxy.BrokerConn0.OffsetRequestCallCount, Is.EqualTo(1));
            Assert.That(routerProxy.BrokerConn1.OffsetRequestCallCount, Is.EqualTo(1));
        }

        public void BrokerRouteShouldCycleThroughEachBrokerUntilOneIsFound()
        {
            var routerProxy = new BrokerRouterProxy(_kernel);
            routerProxy.BrokerConn0.MetadataResponseFunction = () => { throw new Exception("some error"); };
            var router = routerProxy.Create();

            var result = router.GetTopicMetadata(TestTopic);
            Assert.That(result, Is.Not.Null);
            Assert.That(routerProxy.BrokerConn0.MetadataRequestCallCount, Is.EqualTo(1));
            Assert.That(routerProxy.BrokerConn1.MetadataRequestCallCount, Is.EqualTo(1));
        }

        public void GetTopicOffsetShouldThrowAnyException()
        {
            var routerProxy = new BrokerRouterProxy(_kernel);
            routerProxy.BrokerConn0.OffsetResponseFunction = () => { throw new ApplicationException("test 99"); };
            var router = routerProxy.Create();
            var common = new CommonQueries(router);

            common.GetTopicOffsetAsync(BrokerRouterProxy.TestTopic).ContinueWith(t =>
            {
                Assert.That(t.IsFaulted, Is.True);
                Assert.That(t.Exception.Flatten().ToString(), Is.StringContaining("test 99"));
            }).Wait();
        }

        public void BrokerRouteShouldReturnTopicFromCache()
        {
            var routerProxy = new BrokerRouterProxy(_kernel);
            var router = routerProxy.Create();

            var result1 = router.GetTopicMetadata(TestTopic);
            var result2 = router.GetTopicMetadata(TestTopic);

            Assert.That(routerProxy.BrokerConn0.MetadataRequestCallCount, Is.EqualTo(1));
            Assert.That(result1.Count, Is.EqualTo(1));
            Assert.That(result1[0].Name, Is.EqualTo(TestTopic));
            Assert.That(result2.Count, Is.EqualTo(1));
            Assert.That(result2[0].Name, Is.EqualTo(TestTopic));
        }

        public async Task ShouldTryToRefreshMataDataIfCanRecoverByRefreshMetadata(ErrorResponseCode code)
        {
            var routerProxy = new BrokerRouterProxy(_kernel);
            routerProxy._cacheExpiration = new TimeSpan(10);
            var router = routerProxy.Create();
            ProtocolGateway protocolGateway = new ProtocolGateway(router);

            routerProxy.BrokerConn0.FetchResponseFunction = FailedInFirstMessageError(code, routerProxy._cacheExpiration);
            routerProxy.BrokerConn0.MetadataResponseFunction = BrokerRouterProxy.CreateMetadataResponseWithMultipleBrokers;

            await protocolGateway.SendProtocolRequest(new FetchRequest(), BrokerRouterProxy.TestTopic, _partitionId);

            Assert.That(routerProxy.BrokerConn0.MetadataRequestCallCount, Is.EqualTo(2));
            Assert.That(routerProxy.BrokerConn0.FetchRequestCallCount, Is.EqualTo(2));
        }

        public async Task ShouldTryToRefreshMataDataIfOnExceptions(Type exceptionType)
        {
            var routerProxy = new BrokerRouterProxy(_kernel);
            routerProxy._cacheExpiration = TimeSpan.FromMilliseconds(10);
            var router = routerProxy.Create();
            ProtocolGateway protocolGateway = new ProtocolGateway(router);

            routerProxy.BrokerConn0.FetchResponseFunction = FailedInFirstMessageException(exceptionType, routerProxy._cacheExpiration);
            routerProxy.BrokerConn0.MetadataResponseFunction = BrokerRouterProxy.CreateMetadataResponseWithMultipleBrokers;

            await protocolGateway.SendProtocolRequest(new FetchRequest(), BrokerRouterProxy.TestTopic, _partitionId);

            Assert.That(routerProxy.BrokerConn0.MetadataRequestCallCount, Is.EqualTo(2));
            Assert.That(routerProxy.BrokerConn0.FetchRequestCallCount, Is.EqualTo(2));
        }

        public void ConsumerShouldReturnOffset()
        {
            var routerProxy = new BrokerRouterProxy(_kernel);
            routerProxy.BrokerConn0.FetchResponseFunction = () => { return new FetchResponse(); };
            var router = routerProxy.Create();
            var options = CreateOptions(router);
            options.PartitionWhitelist = new List<int>();
            var consumer = new Consumer(options);

            var test = consumer.Consume().Take(1);
            while (consumer.ConsumerTaskCount <= 0)
            {
                Thread.Sleep(100);
            }

            Assert.That(consumer.ConsumerTaskCount, Is.EqualTo(2));
        }

        public void ConsumerWhitelistShouldOnlyConsumeSpecifiedPartition()
        {
            var routerProxy = new BrokerRouterProxy(_kernel);
            routerProxy.BrokerConn0.FetchResponseFunction = () => { return new FetchResponse(); };
            var router = routerProxy.Create();
            var options = CreateOptions(router);
            options.PartitionWhitelist = new List<int> { 0 };
            var consumer = new Consumer(options);

            var test = consumer.Consume().Take(1);
            while (consumer.ConsumerTaskCount <= 0)
            {
                Thread.Sleep(100);
            }

            Assert.That(routerProxy.BrokerConn0.FetchRequestCallCount, Is.GreaterThanOrEqualTo(1));
            Assert.That(routerProxy.BrokerConn1.FetchRequestCallCount, Is.EqualTo(0));
        }

        public void ConsumerWhitelistShouldOnlyConsumeSpecifiedPartition()
        {
            var routerProxy = new BrokerRouterProxy(new MoqMockingKernel());
            routerProxy.BrokerConn0.FetchResponseFunction = () => { return new FetchResponse(); };
            var router = routerProxy.Create();
            var options = CreateOptions(router);
            options.PartitionWhitelist = new List<int> { 0 };
            using (var consumer = new Consumer(options))
            {
                var test = consumer.Consume();

                TaskTest.WaitFor(() => consumer.ConsumerTaskCount > 0);
                TaskTest.WaitFor(() => routerProxy.BrokerConn0.FetchRequestCallCount > 0);

                Assert.That(consumer.ConsumerTaskCount, Is.EqualTo(1), "Consumer should only create one consuming thread for partition 0.");
                Assert.That(routerProxy.BrokerConn0.FetchRequestCallCount, Is.GreaterThanOrEqualTo(1));
                Assert.That(routerProxy.BrokerConn1.FetchRequestCallCount, Is.EqualTo(0));
            }
        }

        public void ConnectionExceptionOnOneShouldCommunicateBackWhichMessagesFailed()
        {
            //TODO is there a way to communicate back which client failed and which succeeded.
            var routerProxy = new BrokerRouterProxy(_kernel);
            routerProxy.BrokerConn1.ProduceResponseFunction = () => { throw new ApplicationException("some exception"); };

            var router = routerProxy.Create();
            var producer = new Producer(router);

            var messages = new List<Message>
                {
                    new Message{Value = "1"}, new Message{Value = "2"}
                };

            //this will produce an exception, but message 1 succeeded and message 2 did not.
            //should we return a ProduceResponse with an error and no error for the other messages?
            //at this point though the client does not know which message is routed to which server.
            //the whole batch of messages would need to be returned.
            var test = producer.SendMessageAsync("UnitTest", messages).Result;
        }

        public void ConsumerWithEmptyWhitelistShouldConsumeAllPartition()
        {
            var routerProxy = new BrokerRouterProxy(new MoqMockingKernel());

            var router = routerProxy.Create();
            var options = CreateOptions(router);
            options.PartitionWhitelist = new List<int>();

            using (var consumer = new Consumer(options))
            {
                var test = consumer.Consume();

                TaskTest.WaitFor(() => consumer.ConsumerTaskCount > 0);
                TaskTest.WaitFor(() => routerProxy.BrokerConn0.FetchRequestCallCount > 0);
                TaskTest.WaitFor(() => routerProxy.BrokerConn1.FetchRequestCallCount > 0);

                Assert.That(consumer.ConsumerTaskCount, Is.EqualTo(2), "Consumer should create one consuming thread for each partition.");
                Assert.That(routerProxy.BrokerConn0.FetchRequestCallCount, Is.GreaterThanOrEqualTo(1), "BrokerConn0 not sent FetchRequest");
                Assert.That(routerProxy.BrokerConn1.FetchRequestCallCount, Is.GreaterThanOrEqualTo(1), "BrokerConn1 not sent FetchRequest");
            }
        }

        public void CancellationShouldInterruptConsumption()
        {
            var routerProxy = new BrokerRouterProxy(_kernel);
            routerProxy.BrokerConn0.FetchResponseFunction = () => { while (true) Thread.Yield(); };

            var router = routerProxy.Create();

            var options = CreateOptions(router);

            var consumer = new Consumer(options);

            var tokenSrc = new CancellationTokenSource();

            var consumeTask = Task.Run(() => consumer.Consume(tokenSrc.Token).FirstOrDefault());

            if (consumeTask.Wait(TimeSpan.FromSeconds(3)))
                Assert.Fail();

            tokenSrc.Cancel();

            Assert.That(
                Assert.Throws<AggregateException>(consumeTask.Wait).InnerException,
                Is.TypeOf<OperationCanceledException>());
        }

        public void SelectPartitionShouldUsePartitionSelector(string testCase)
        {
            var key = testCase.ToIntSizedBytes();
            var routerProxy = new BrokerRouterProxy(_kernel);

            _mockPartitionSelector.Setup(x => x.Select(It.IsAny<Topic>(), key))
                                  .Returns(() => new Partition
                                  {
                                      ErrorCode = 0,
                                      Isrs = new List<int> { 1 },
                                      PartitionId = 0,
                                      LeaderId = 0,
                                      Replicas = new List<int> { 1 },
                                  });

            routerProxy.PartitionSelector = _mockPartitionSelector.Object;

            var result = routerProxy.Create().SelectBrokerRoute(TestTopic, key);

            _mockPartitionSelector.Verify(f => f.Select(It.Is<Topic>(x => x.Name == TestTopic), key), Times.Once());
        }

        public async Task SendProtocolRequestShouldThrowException(Type exceptionType)
        {
            var routerProxy = new BrokerRouterProxy(_kernel);
            routerProxy._cacheExpiration = TimeSpan.FromMilliseconds(10);
            var router = routerProxy.Create();
            ProtocolGateway protocolGateway = new ProtocolGateway(router);

            routerProxy.BrokerConn0.FetchResponseFunction = FailedInFirstMessageException(exceptionType, routerProxy._cacheExpiration);
            routerProxy.BrokerConn0.MetadataResponseFunction = BrokerRouterProxy.CreateMetadataResponseWithMultipleBrokers;
            try
            {
                await protocolGateway.SendProtocolRequest(new FetchRequest(), BrokerRouterProxy.TestTopic, _partitionId);
                Assert.IsTrue(false, "Should throw exception");
            }
            catch (Exception ex)
            {
                Assert.That(ex.GetType(), Is.EqualTo(exceptionType));
            }
        }

        public async Task SendProtocolRequestShouldThrowException(Type exceptionType)
        {
            var routerProxy = new BrokerRouterProxy(_kernel);
            routerProxy._cacheExpiration = TimeSpan.FromMilliseconds(10);
            var router = routerProxy.Create();

            int partitionId = 0;
            ProtocolGateway protocolGateway = new ProtocolGateway(router);
            var fetchRequest = new FetchRequest();

            bool firstTime = true;
            Func<Task<FetchResponse>> ShouldReturnError = async () =>
            {
                if (firstTime)
                {
                    firstTime = !firstTime;
                    object[] args = new object[1];
                    args[0] = "error Test";
                    throw (Exception)Activator.CreateInstance(exceptionType, args);
                }
                return new FetchResponse() { Error = (short)ErrorResponseCode.NoError };
            };

            routerProxy.BrokerConn0.FetchResponseFunction = ShouldReturnError;
            routerProxy.BrokerConn0.MetadataResponseFunction = BrokerRouterProxy.DefaultMetadataResponse;
            try
            {
                await protocolGateway.SendProtocolRequest(fetchRequest, BrokerRouterProxy.TestTopic, partitionId);
                Assert.IsTrue(false, "Should throw exception");
            }
            catch (Exception ex)
            {
                Assert.That(ex.GetType(), Is.EqualTo(exceptionType));
            }
        }