[chore] [receiver/solacereceiver] Fixed a few flaky tests in receiver…

…/solacereceiver (open-telemetry#20348)

Addresses a couple of test failures with two main problems:

- Addressed an issue where on windows the timer window was too small to get within a scheduling tick. Upped the timeout in this case.
- Addressed a cache coherency issue with callbacks on mock transports. Replaced callbacks with atomic unsafe.Pointers.

receiver/solacereceiver/messaging_service_test.go

@@ -22,8 +22,10 @@ import (
  "net"
  "reflect"
  "runtime"
+ "sync/atomic"
  "testing"
  "time"
+ "unsafe"
 
  "github.com/Azure/go-amqp"
  "github.com/stretchr/testify/assert"
@@ -283,10 +285,10 @@ func TestAMQPNewClientDialAndCloseCtxTimeoutFailure(t *testing.T) {
  service, conn := startMockedService(t)
 
  closed := false
- conn.closeHandle = func() error {
+ conn.setCloseHandler(func() error {
  closed = true
  return nil
- }
+ })
  ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
  defer cancel()
  go func() {
@@ -309,10 +311,10 @@ func TestAMQPNewClientDialAndCloseConnFailure(t *testing.T) {
  mockWriteData(conn, writeData)
 
  closed := false
- conn.closeHandle = func() error {
+ conn.setCloseHandler(func() error {
  closed = true
  return fmt.Errorf("some error")
- }
+ })
  service.close(context.Background())
  // expect conn.Close to have been called
  assert.True(t, closed)
@@ -345,13 +347,13 @@ func TestAMQPAcknowledgeMessage(t *testing.T) {
  writeCalled := make(chan struct{})
  // Expected accept from AMQP frame for first received message
  // "\x00\x00\x00\x1c\x02\x00\x00\x00\x00\x53\x15\xd0\x00\x00\x00\x0c\x00\x00\x00\x05\x41\x43\x40\x41\x00\x53\x24\x45"
- conn.writeHandle = func(b []byte) (n int, err error) {
+ conn.setWriteHandler(func(b []byte) (n int, err error) {
  // assert that a disposition is written
  assert.Equal(t, byte(0x15), b[10])
  assert.Equal(t, byte(0x24), b[26]) // 0x24 at the 27th byte in this case means accept
  close(writeCalled)
  return len(b), nil
- }
+ })
  err = service.accept(context.Background(), msg)
  assert.NoError(t, err)
  assertChannelClosed(t, writeCalled)
@@ -366,13 +368,13 @@ func TestAMQPModifyMessage(t *testing.T) {
  writeCalled := make(chan struct{})
  // Expected modify from AMQP frame for first received message
  // "\x00\x00\x00\x1c\x02\x00\x00\x00\x00\x53\x15\xd0\x00\x00\x00\x0c\x00\x00\x00\x05\x41\x43\x40\x41\x00\x53\x25\x45"
- conn.writeHandle = func(b []byte) (n int, err error) {
+ conn.setWriteHandler(func(b []byte) (n int, err error) {
  // assert that a disposition is written
  assert.Equal(t, byte(0x15), b[10])
  assert.Equal(t, byte(0x27), b[26]) // 0x27 at the 27th byte in this case means modify
  close(writeCalled)
  return len(b), nil
- }
+ })
  err = service.failed(context.Background(), msg)
  assert.NoError(t, err)
  select {
@@ -420,10 +422,10 @@ func closeMockedAMQPService(t *testing.T, service *amqpMessagingService, conn *c
  mockWriteData(conn, writeData)
 
  closed := false
- conn.closeHandle = func() error {
+ conn.setCloseHandler(func() error {
  closed = true
  return nil
- }
+ })
  service.close(context.Background())
  // expect conn.Close to have been called
  assert.True(t, closed)
@@ -480,7 +482,7 @@ func TestAMQPNewClientDialWithBadAttachResponseExpectingError(t *testing.T) {
 }
 
 func mockWriteData(conn *connMock, data [][]byte, callbacks ...func(sentData, receivedData []byte)) {
- conn.writeHandle = func(b []byte) (n int, err error) {
+ conn.setWriteHandler(func(b []byte) (n int, err error) {
  var next []byte
  if len(data) != 0 {
  next = data[0]
@@ -493,7 +495,7 @@ func mockWriteData(conn *connMock, data [][]byte, callbacks ...func(sentData, re
  conn.nextData <- next
  }
  return len(b), nil
- }
+ })
 }
 
 func mockDialFunc(conn *connMock) {
@@ -640,11 +642,22 @@ func TestConfigAMQPAuthenticationNoDetails(t *testing.T) {
 // Write will call writeHandle
 type connMock struct {
  nextData chan []byte
- writeHandle func([]byte) (n int, err error)
- closeHandle func() error
+ writeHandle unsafe.Pointer // expected type: func([]byte) (n int, err error)
+ closeHandle unsafe.Pointer // expected type: func() error
  remaining *bytes.Reader
 }
 
+type writeHandler func([]byte) (n int, err error)
+type closeHandler func() error
+
+func (c *connMock) setWriteHandler(handler writeHandler) {
+ atomic.StorePointer(&c.writeHandle, unsafe.Pointer(&handler))
+}
+
+func (c *connMock) setCloseHandler(handler closeHandler) {
+ atomic.StorePointer(&c.closeHandle, unsafe.Pointer(&handler))
+}
+
 func (c *connMock) Read(b []byte) (n int, err error) {
  if c.remaining == nil {
  d := <-c.nextData
@@ -663,15 +676,17 @@ func (c *connMock) Read(b []byte) (n int, err error) {
 }
 
 func (c *connMock) Write(b []byte) (n int, err error) {
- if c.writeHandle != nil {
- return c.writeHandle(b)
+ handlerPointer := atomic.LoadPointer(&c.writeHandle)
+ if handlerPointer != nil {
+ return (*(*writeHandler)(handlerPointer))(b)
  }
  return len(b), nil
 }
 
 func (c *connMock) Close() error {
- if c.closeHandle != nil {
- return c.closeHandle()
+ handlerPointer := atomic.LoadPointer(&c.closeHandle)
+ if handlerPointer != nil {
+ return (*(*closeHandler)(handlerPointer))()
  }
  return nil
 }

receiver/solacereceiver/receiver_test.go

@@ -352,7 +352,7 @@ func TestReceiverFlowControlDelayedRetry(t *testing.T) {
  for _, tc := range testCases {
  t.Run(tc.name, func(t *testing.T) {
  receiver, messagingService, unmarshaller := newReceiver(t)
- delay := 5 * time.Millisecond
+ delay := 50 * time.Millisecond
  // Increase delay on windows due to tick granularity
  // https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/17197
  if runtime.GOOS == "windows" {
@@ -461,7 +461,12 @@ func TestReceiverFlowControlDelayedRetryInterrupt(t *testing.T) {
 func TestReceiverFlowControlDelayedRetryMultipleRetries(t *testing.T) {
  receiver, messagingService, unmarshaller := newReceiver(t)
  // we won't wait 10 seconds since we will interrupt well before
- retryInterval := 20 * time.Millisecond
+ retryInterval := 50 * time.Millisecond
+ // Increase delay on windows due to tick granularity
+ // https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/19409
+ if runtime.GOOS == "windows" {
+ retryInterval = 500 * time.Millisecond
+ }
  var retryCount int64 = 5
  receiver.config.Flow.DelayedRetry.Delay = retryInterval
  var err error