feat(test): add per-partition isolation integration tests for subscriber and consumer

Add integration tests that verify per-partition isolation at both the
subscriber and consumer layers, ensuring a slow partition doesn't block
others.

Subscriber-level (existing suite):
- TestSubscriberPerPartitionIsolation: holds one partition's ack, verifies
  the other partition's delivery still arrives promptly
- TestSubscriberPartitionOrderPreserved: publishes 5 messages to one
  partition, asserts FIFO delivery order

Consumer-level (new suite at test/integration/core/consumer/):
- TestConsumerPerPartitionIsolation: blocks controller on partition-a,
  verifies partition-b processes independently
- TestConsumerPartitionOrdering: verifies single-partition FIFO at the
  consumer/controller layer
- TestConsumerMultiPartitionThroughput: verifies 3 partitions process
  concurrently (wall-clock < serial time)

Also adds consumer-integration-test CI job and Makefile target.

Author: behinddwalls

test/integration/core/consumer/BUILD.bazel

@@ -15,7 +15,6 @@ go_test(
         "//extension/queue/mysql",
         "//test/testutil",
         "@com_github_go_sql_driver_mysql//:mysql",
-        "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",
         "@com_github_stretchr_testify//suite",
         "@com_github_uber_go_tally_v4//:tally",

test/integration/core/consumer/consumer_test.go

@@ -9,7 +9,6 @@ import (
 	"time"
 
 	_ "github.com/go-sql-driver/mysql"
-	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 	"github.com/uber-go/tally/v4"
@@ -50,6 +49,9 @@ func (c *testController) ConsumerGroup() string {
 // testTimeout is the safety-net duration for channel waits in integration tests.
 const testTimeout = 10 * time.Second
 
+// stopTimeoutMs is the timeout in milliseconds for consumer.Stop().
+const stopTimeoutMs = 10000
+
 type ConsumerIntegrationSuite struct {
 	suite.Suite
 	ctx   context.Context
@@ -199,7 +201,7 @@ func (s *ConsumerIntegrationSuite) TestConsumerPerPartitionIsolation() {
 	case <-partAStarted:
 		s.log.Logf("partition-a processing started (blocking)")
 	case <-time.After(testTimeout):
-		t.Fatal("Timeout waiting for partition-a to start processing")
+		require.FailNow(t, "Timeout waiting for partition-a to start processing")
 	}
 
 	// Now publish to partition-b — should be processed even though partition-a is blocked
@@ -210,12 +212,12 @@ func (s *ConsumerIntegrationSuite) TestConsumerPerPartitionIsolation() {
 	case <-partBProcessed:
 		s.log.Logf("partition-b processed while partition-a was blocked")
 	case <-time.After(testTimeout):
-		t.Fatal("Timeout waiting for partition-b: partition-a blocked it (no isolation)")
+		require.FailNow(t, "Timeout waiting for partition-b: partition-a blocked it (no isolation)")
 	}
 
 	// Unblock partition-a and stop cleanly
 	close(partAUnblock)
-	require.NoError(t, c.Stop(10000))
+	require.NoError(t, c.Stop(stopTimeoutMs))
 
 	s.log.Logf("Per-partition isolation verified at consumer level")
 }
@@ -277,17 +279,17 @@ func (s *ConsumerIntegrationSuite) TestConsumerPartitionOrdering() {
 	select {
 	case <-allDone:
 	case <-time.After(3 * testTimeout):
-		t.Fatal("Timeout waiting for all messages to be processed")
+		require.FailNow(t, "Timeout waiting for all messages to be processed")
 	}
 
-	require.NoError(t, c.Stop(10000))
+	require.NoError(t, c.Stop(stopTimeoutMs))
 
 	// Assert order matches publish order
 	mu.Lock()
 	defer mu.Unlock()
 	require.Len(t, processedIDs, numMessages)
 	for i := range numMessages {
-		assert.Equal(t, publishedIDs[i], processedIDs[i],
+		require.Equal(t, publishedIDs[i], processedIDs[i],
 			"Message at position %d out of order: expected %s, got %s",
 			i, publishedIDs[i], processedIDs[i])
 	}
@@ -351,10 +353,10 @@ func (s *ConsumerIntegrationSuite) TestConsumerMultiPartitionThroughput() {
 	select {
 	case <-allDone:
 	case <-time.After(3 * testTimeout):
-		t.Fatal("Timeout waiting for all partitions to be processed — partitions may not be running in parallel")
+		require.FailNow(t, "Timeout waiting for all partitions to be processed — partitions may not be running in parallel")
 	}
 
-	require.NoError(t, c.Stop(10000))
+	require.NoError(t, c.Stop(stopTimeoutMs))
 
 	s.log.Logf("Multi-partition throughput verified: %d partitions processed concurrently", numPartitions)
 }

test/integration/extension/queue/mysql/queue_test.go

@@ -23,6 +23,9 @@ import (
 	"github.com/uber/submitqueue/test/testutil"
 )
 
+// testTimeout is the safety-net duration for channel waits in integration tests.
+const testTimeout = 10 * time.Second
+
 type SQLQueueIntegrationSuite struct {
 	suite.Suite
 	ctx   context.Context
@@ -279,7 +282,7 @@ func (s *SQLQueueIntegrationSuite) TestSubscriberPartitionOrderPreserved() {
 	require.NoError(t, err)
 
 	receivedIDs := make([]string, 0, numMessages)
-	receiveNWithTimeout(t, deliveryChan, numMessages, 10*time.Second, func(delivery extqueue.Delivery, index int) {
+	receiveNWithTimeout(t, deliveryChan, numMessages, testTimeout, func(delivery extqueue.Delivery, index int) {
 		msgID := delivery.Message().ID
 		receivedIDs = append(receivedIDs, msgID)
 		t.Logf("Received: %s", msgID)