Fix flaky test with DataWriter (part 2)

[grodowski]

Improves the DataWriter setup by creating threads and connecting earlier in initialize. Includes synchronization on DataWriter threads, so that calling #start ensures a first write to db before returning and letting the test proceed.

This attempts to improve flakiness of test_interrupt_resume_inline_verifier_with_datawriter, which sometimes fails as expected BinlogVerifyStore contents are empty on interrup, when they should contain some data from DataWriter.

I found out with the below log entry that nothing gets to the inline verifier before interrupt in cases where test_interrupt_resume_inline_verifier_with_datawriter fails (feel free to try out locally).

Hoping this will be the definitive (and correct) fix, as opposed to the speed-up from #371.

---

diff --git a/inline_verifier.go b/inline_verifier.go
index 552c88e..b1ae817 100644
--- a/inline_verifier.go
+++ b/inline_verifier.go
@@ -25,6 +25,7 @@ import (
 // TODO: remove IterativeVerifier and remove this comment.
 type BinlogVerifyStore struct {
 	EmitLogPerRowsAdded uint64
+	logger              *logrus.Entry
 
 	mutex *sync.Mutex
 	// db => table => paginationKey => number of times it changed.
@@ -110,6 +111,7 @@ func NewBinlogVerifyStore() *BinlogVerifyStore {
 		store:               make(map[string]map[string]map[uint64]int),
 		totalRowCount:       uint64(0),
 		currentRowCount:     uint64(0),
+		logger:              logrus.WithField("tag", "BinLogVerifier"),
 	}
 }
 
@@ -127,6 +129,8 @@ func (s *BinlogVerifyStore) Add(table *TableSchema, paginationKey uint64) {
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
 
+	s.logger.Error("Add event to store DEBUG")
+
 	_, exists := s.store[table.Schema]
 	if !exists {
 		s.store[table.Schema] = make(map[string]map[uint64]int)

[grodowski]

This is still a problem 👀

[grodowski]

Lets #start return after first write

[mtaner]

we should still close the connection after the stop is requested?

[grodowski]

Yup, I forgot to add it back here after an experiment with the ConnectionPool gem (but thought it's a an overkill for this test setup)

[mtaner]

nit: this ensure will only get called if something goes wrong in begin block right? So if anything goes wrong before that we won't close the connection.

[grodowski]

I don't think that's right, ensures are always called after any rescues or at the end of the block. Just moved it from here

[mtaner]

when I do this:

def test
  raise "here"
   
   begin
     puts "Hi"
   ensure
     puts "always ensure"
   end
 end

ensure is not called.

irb(main):010> test
(irb):2:in `test': here (RuntimeError)

but when I do this:

def test
  puts "here"
   
   begin
     raise "Hi"
   ensure
     puts "always ensure"
   end
 end

I get this:

irb(main):020> test
here
always ensure
(irb):15:in `test': Hi (RuntimeError)

[grodowski]

Ohhh right, just saw what happened there as I moved the connection outside of the block. Simplified and fixed in 50c78b2

[grodowski]

Currently re-running the CI suite several times to evaluate its stability ⌛ I don't think it's going to resolve all concurrency issues, but hopefully InterruptResumeTest will be more consistent

Edit: 96f92f5 is a side-quest for another go-test flake (where synchronization would be ideal too...)