[FAB-2915] Set 'last block cut' to proper value

https://jira.hyperledger.org/browse/FAB-2915

Both the "last persisted offset" and the "last cut block" fields need to
be set to their proper values before a restarted chain can resume
operation. We were taking care of the former, but not the latter. This
changeset fixes that.

Change-Id: If688947d49566c88b416d7abb6f0e5b8aa42d977
Signed-off-by: Kostas Christidis <kostas@christidis.io>

orderer/kafka/orderer.go

@@ -105,13 +105,16 @@ func getLastOffsetPersisted(metadata *cb.Metadata, chainID string) int64 {
 // be satisfied by both the actual and the mock object and will allow
 // us to retrieve these constructors.
 func newChain(consenter testableConsenter, support multichain.ConsenterSupport, lastOffsetPersisted int64) *chainImpl {
-	logger.Debugf("[channel: %s] Starting chain with last persisted offset: %d", support.ChainID(), lastOffsetPersisted)
+	lastCutBlock := support.Height() - 1
+	logger.Debugf("[channel: %s] Starting chain with last persisted offset %d and last recorded block %d",
+		support.ChainID(), lastOffsetPersisted, lastCutBlock)
 	return &chainImpl{
 		consenter:           consenter,
 		support:             support,
 		partition:           newChainPartition(support.ChainID(), rawPartition),
 		batchTimeout:        support.SharedConfig().BatchTimeout(),
 		lastOffsetPersisted: lastOffsetPersisted,
+		lastCutBlock:        lastCutBlock,
 		producer:            consenter.prodFunc()(support.SharedConfig().KafkaBrokers(), consenter.kafkaVersion(), consenter.retryOptions(), consenter.tlsConfig()),
 		halted:              false, // Redundant as the default value for booleans is false but added for readability
 		exitChan:            make(chan struct{}),

orderer/mocks/multichain/multichain.go

@@ -45,6 +45,9 @@ type ConsenterSupport struct {
 	// ChainIDVal is the value returned by ChainID()
 	ChainIDVal string
 
+	// HeightVal is the value returned by Height()
+	HeightVal uint64
+
 	// NextBlockVal stores the block created by the most recent CreateNextBlock() call
 	NextBlockVal *cb.Block
 
@@ -83,6 +86,7 @@ func (mcs *ConsenterSupport) WriteBlock(block *cb.Block, _committers []filter.Co
 		umtxs[i] = utils.UnmarshalEnvelopeOrPanic(block.Data.Data[i])
 	}
 	mcs.Batches <- umtxs
+	mcs.HeightVal++
 	if encodedMetadataValue != nil {
 		block.Metadata.Metadata[cb.BlockMetadataIndex_ORDERER] = utils.MarshalOrPanic(&cb.Metadata{Value: encodedMetadataValue})
 	}
@@ -95,6 +99,11 @@ func (mcs *ConsenterSupport) ChainID() string {
 	return mcs.ChainIDVal
 }
 
+// Height returns the number of blocks of the chain this specific consenter instance is associated with
+func (mcs *ConsenterSupport) Height() uint64 {
+	return mcs.HeightVal
+}
+
 // Sign returns the bytes passed in
 func (mcs *ConsenterSupport) Sign(message []byte) ([]byte, error) {
 	return message, nil

orderer/multichain/chainsupport.go

@@ -70,6 +70,7 @@ type ConsenterSupport interface {
 	CreateNextBlock(messages []*cb.Envelope) *cb.Block
 	WriteBlock(block *cb.Block, committers []filter.Committer, encodedMetadataValue []byte) *cb.Block
 	ChainID() string // ChainID returns the chain ID this specific consenter instance is associated with
+	Height() uint64  // Returns the number of blocks on the chain this specific consenter instance is associated with
 }
 
 // ChainSupport provides a wrapper for the resources backing a chain
@@ -258,3 +259,7 @@ func (cs *chainSupport) WriteBlock(block *cb.Block, committers []filter.Committe
 	}
 	return block
 }
+
+func (cs *chainSupport) Height() uint64 {
+	return cs.Reader().Height()
+}