Merge branch 'master' into refactor/remove-mobx-from-quotas-page

Author: jvorcak

.github/workflows/repository-dispatch.yml

@@ -8,6 +8,7 @@ on:
       - '.github/workflows/repository-dispatch.yml'
       - '.github/workflows/backend-lint-test.yml'
       - '.github/workflows/frontend-verify.yml'
+      - 'test-images.json'
     tags:
       - '*'
     branches:

backend/pkg/api/connect/service/console/service.go

@@ -61,6 +61,7 @@ func (api *Service) ListMessages(
 		MaxResults:            int(req.Msg.GetMaxResults()),
 		FilterInterpreterCode: req.Msg.GetFilterInterpreterCode(),
 		Enterprise:            req.Msg.GetEnterprise(),
+		PageToken:             req.Msg.GetPageToken(),
 	}
 
 	interpreterCode, err := lmq.DecodeInterpreterCode()
@@ -96,6 +97,8 @@ func (api *Service) ListMessages(
 		IgnoreMaxSizeLimit:    req.Msg.GetIgnoreMaxSizeLimit(),
 		KeyDeserializer:       fromProtoEncoding(req.Msg.GetKeyDeserializer()),
 		ValueDeserializer:     fromProtoEncoding(req.Msg.GetValueDeserializer()),
+		PageToken:             lmq.PageToken,
+		PageSize:              int(req.Msg.GetPageSize()),
 	}
 
 	timeout := 35 * time.Second

backend/pkg/api/connect/service/console/stream_progress_reporter.go

@@ -213,7 +213,7 @@ func (p *streamProgressReporter) OnMessage(message *console.TopicMessage) {
 	}
 }
 
-func (p *streamProgressReporter) OnComplete(elapsedMs int64, isCancelled bool) {
+func (p *streamProgressReporter) OnComplete(elapsedMs int64, isCancelled bool, nextPageToken string) {
 	p.writeMutex.Lock()
 	defer p.writeMutex.Unlock()
 
@@ -222,6 +222,7 @@ func (p *streamProgressReporter) OnComplete(elapsedMs int64, isCancelled bool) {
 		IsCancelled:      isCancelled,
 		MessagesConsumed: p.messagesConsumed.Load(),
 		BytesConsumed:    p.bytesConsumed.Load(),
+		NextPageToken:    nextPageToken,
 	}
 
 	if err := p.stream.Send(

backend/pkg/api/handle_topic_messages_integration_test.go

@@ -28,6 +28,7 @@ import (
 	"connectrpc.com/connect"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"github.com/twmb/franz-go/pkg/kadm"
 	"github.com/twmb/franz-go/pkg/kgo"
 	"github.com/twmb/franz-go/pkg/sr"
 	"google.golang.org/genproto/googleapis/rpc/errdetails"
@@ -1050,3 +1051,222 @@ func randomString(n int) string {
 	}
 	return string(b)
 }
+
+func (s *APIIntegrationTestSuite) TestListMessages_Pagination() {
+	t := s.T()
+
+	require := require.New(t)
+	assert := assert.New(t)
+
+	ctx := t.Context()
+
+	client := v1ac.NewConsoleServiceClient(
+		http.DefaultClient,
+		s.httpAddress(),
+		connect.WithGRPCWeb(),
+	)
+
+	// Create test topic with multiple messages
+	testTopicName := testutil.TopicNameForTest("pagination_test")
+	// Create topic
+	_, err := s.kafkaAdminClient.CreateTopic(ctx, 3, 1, nil, testTopicName)
+	require.NoError(err)
+
+	defer func() {
+		s.kafkaAdminClient.DeleteTopics(ctx, testTopicName)
+	}()
+
+	// Wait for topic to be ready
+	require.Eventually(func() bool {
+		metadata, err := s.kafkaAdminClient.Metadata(ctx, testTopicName)
+		if err != nil {
+			return false
+		}
+		topic, exists := metadata.Topics[testTopicName]
+		return exists && topic.Err == nil && len(topic.Partitions) == 3
+	}, 30*time.Second, 100*time.Millisecond, "Topic should be created and ready")
+
+	// Produce messages
+	const messageCount = 150
+	records := make([]*kgo.Record, messageCount)
+	for i := range messageCount {
+		records[i] = &kgo.Record{
+			Key:   []byte(fmt.Sprintf("key-%d", i)),
+			Value: []byte(fmt.Sprintf(`{"id": %d, "message": "test message %d"}`, i, i)),
+			Topic: testTopicName,
+		}
+	}
+
+	produceResults := s.kafkaClient.ProduceSync(ctx, records...)
+	require.NoError(produceResults.FirstErr())
+
+	// Wait for all produced messages to be committed
+	require.Eventually(func() bool {
+		offsets, err := s.kafkaAdminClient.ListEndOffsets(ctx, testTopicName)
+		if err != nil {
+			return false
+		}
+		totalMessages := int64(0)
+		offsets.Each(func(offset kadm.ListedOffset) {
+			totalMessages += offset.Offset
+		})
+		t.Logf("Total messages across all partitions: %d (expected %d)", totalMessages, messageCount)
+		return totalMessages >= int64(messageCount)
+	}, 30*time.Second, 100*time.Millisecond, "All produced messages should be committed")
+
+	t.Run("first page with cursor pagination", func(t *testing.T) {
+		stream, err := client.ListMessages(ctx, connect.NewRequest(&v1pb.ListMessagesRequest{
+			Topic:       testTopicName,
+			PartitionId: -1, // All partitions
+			StartOffset: -1, // Recent
+			PageSize:    50, // Triggers cursor pagination
+			PageToken:   "", // First page
+		}))
+		require.NoError(err)
+		require.NotNil(stream)
+
+		var messages []*v1pb.ListMessagesResponse_DataMessage
+		var done *v1pb.ListMessagesResponse_StreamCompletedMessage
+
+		for stream.Receive() {
+			msg := stream.Msg()
+			switch v := msg.ControlMessage.(type) {
+			case *v1pb.ListMessagesResponse_Data:
+				messages = append(messages, v.Data)
+			case *v1pb.ListMessagesResponse_Done:
+				done = v.Done
+			}
+		}
+
+		require.NoError(stream.Err())
+		require.NotNil(done, "Should have completion message")
+
+		// Verify first page results
+		assert.Equal(int64(50), done.MessagesConsumed, "Should consume 50 messages (default page size)")
+		assert.NotEmpty(done.NextPageToken, "Should have next page token")
+		assert.Len(messages, 50, "Should return 50 messages")
+
+		// Verify descending order within partitions
+		// Group messages by partition
+		messagesByPartition := make(map[int32][]*v1pb.ListMessagesResponse_DataMessage)
+		for _, msg := range messages {
+			messagesByPartition[msg.PartitionId] = append(messagesByPartition[msg.PartitionId], msg)
+		}
+
+		// Check each partition is in descending order
+		for partitionID, partitionMessages := range messagesByPartition {
+			if len(partitionMessages) > 1 {
+				for i := 0; i < len(partitionMessages)-1; i++ {
+					assert.Greater(partitionMessages[i].Offset, partitionMessages[i+1].Offset,
+						"Messages in partition %d should be in descending offset order", partitionID)
+				}
+			}
+		}
+	})
+
+	t.Run("paginate through all messages", func(t *testing.T) {
+		var allMessages []*v1pb.ListMessagesResponse_DataMessage
+		pageToken := ""
+		pageCount := 0
+		maxPages := 5 // Safety limit
+
+		for pageCount < maxPages {
+			stream, err := client.ListMessages(ctx, connect.NewRequest(&v1pb.ListMessagesRequest{
+				Topic:       testTopicName,
+				PartitionId: -1,
+				StartOffset: -1,
+				PageSize:    50,
+				PageToken:   pageToken,
+			}))
+			require.NoError(err)
+			require.NotNil(stream)
+
+			var pageMessages []*v1pb.ListMessagesResponse_DataMessage
+			var done *v1pb.ListMessagesResponse_StreamCompletedMessage
+
+			for stream.Receive() {
+				msg := stream.Msg()
+				switch v := msg.ControlMessage.(type) {
+				case *v1pb.ListMessagesResponse_Data:
+					pageMessages = append(pageMessages, v.Data)
+				case *v1pb.ListMessagesResponse_Done:
+					done = v.Done
+				}
+			}
+
+			require.NoError(stream.Err())
+			require.NotNil(done)
+
+			allMessages = append(allMessages, pageMessages...)
+			pageCount++
+
+			t.Logf("Page %d: fetched %d messages, nextPageToken=%s", pageCount, len(pageMessages), done.NextPageToken)
+
+			if done.NextPageToken == "" {
+				break
+			}
+
+			pageToken = done.NextPageToken
+		}
+
+		// Verify we got all messages
+		assert.Equal(messageCount, len(allMessages), "Should fetch all %d messages across pages", messageCount)
+		assert.LessOrEqual(pageCount, 4, "Should complete in 3-4 pages (150 messages / 50 per page)")
+	})
+
+	t.Run("error when filter with pagination", func(t *testing.T) {
+		filterCode := base64.StdEncoding.EncodeToString([]byte("return true"))
+
+		stream, err := client.ListMessages(ctx, connect.NewRequest(&v1pb.ListMessagesRequest{
+			Topic:                 testTopicName,
+			PartitionId:           -1,
+			StartOffset:           -1,
+			PageSize:              50,
+			PageToken:             "",
+			FilterInterpreterCode: filterCode,
+		}))
+		require.NoError(err)
+		require.NotNil(stream)
+
+		// Stream should error
+		for stream.Receive() {
+			// Should not receive any messages
+		}
+
+		err = stream.Err()
+		require.Error(err, "Should error when using filter with pagination")
+		assert.Contains(err.Error(), "cannot use filters with pagination")
+	})
+
+	t.Run("standard pagination still works", func(t *testing.T) {
+		stream, err := client.ListMessages(ctx, connect.NewRequest(&v1pb.ListMessagesRequest{
+			Topic:       testTopicName,
+			PartitionId: -1,
+			StartOffset: -2, // Oldest
+			MaxResults:  50, // Standard pagination
+			PageToken:   "",
+		}))
+		require.NoError(err)
+		require.NotNil(stream)
+
+		var messages []*v1pb.ListMessagesResponse_DataMessage
+		var done *v1pb.ListMessagesResponse_StreamCompletedMessage
+
+		for stream.Receive() {
+			msg := stream.Msg()
+			switch v := msg.ControlMessage.(type) {
+			case *v1pb.ListMessagesResponse_Data:
+				messages = append(messages, v.Data)
+			case *v1pb.ListMessagesResponse_Done:
+				done = v.Done
+			}
+		}
+
+		require.NoError(stream.Err())
+		require.NotNil(done)
+
+		// In standard pagination, should not have cursor pagination tokens
+		assert.Empty(done.NextPageToken, "Standard pagination should not return page token")
+		assert.LessOrEqual(len(messages), 50, "Should respect maxResults limit")
+	})
+}

backend/pkg/api/httptypes/types.go

@@ -21,12 +21,16 @@ import (
 // used in Console Enterprise to implement the hooks.
 type ListMessagesRequest struct {
 	TopicName             string `json:"topicName"`
-	StartOffset           int64  `json:"startOffset"`    // -1 for recent (newest - results), -2 for oldest offset, -3 for newest, -4 for timestamp
-	StartTimestamp        int64  `json:"startTimestamp"` // Start offset by unix timestamp in ms (only considered if start offset is set to -4)
-	PartitionID           int32  `json:"partitionId"`    // -1 for all partition ids
-	MaxResults            int    `json:"maxResults"`
+	StartOffset           int64  `json:"startOffset"`           // -1 for recent (newest - results), -2 for oldest offset, -3 for newest, -4 for timestamp
+	StartTimestamp        int64  `json:"startTimestamp"`        // Start offset by unix timestamp in ms (only considered if start offset is set to -4)
+	PartitionID           int32  `json:"partitionId"`           // -1 for all partition ids
+	MaxResults            int    `json:"maxResults"`            // Maximum number of messages to fetch (1-10000).
 	FilterInterpreterCode string `json:"filterInterpreterCode"` // Base64 encoded code
 
+	// Pagination fields (used when PageSize > 0)
+	PageToken string `json:"pageToken,omitempty"`
+	PageSize  int    `json:"pageSize,omitempty"` // Page size for pagination (1-10000)
+
 	// Enterprise may only be set in the Enterprise mode. The JSON deserialization is deferred
 	// to the enterprise backend.
 	Enterprise json.RawMessage `json:"enterprise,omitempty"`
@@ -50,6 +54,15 @@ func (l *ListMessagesRequest) OK() error {
 		return errors.New("max results must be between 1 and 500")
 	}
 
+	// Pagination mode: when PageSize > 0, filters are not supported
+	if l.PageSize > 0 {
+		if l.FilterInterpreterCode != "" {
+			return errors.New("cannot use filters with pagination")
+		}
+		// PageToken validation is done in the console package decodePageToken
+		return nil
+	}
+
 	if _, err := l.DecodeInterpreterCode(); err != nil {
 		return fmt.Errorf("failed to decode interpreter code %w", err)
 	}

backend/pkg/connector/guide/iceberg_sink.go

@@ -22,7 +22,7 @@ func NewIcebergSinkGuide(opts ...Option) Guide {
 		DefaultGuide: DefaultGuide{
 			options: o,
 		},
-		className: "io.tabular.iceberg.connect.IcebergSinkConnector",
+		className: "org.apache.iceberg.connect.IcebergSinkConnector",
 		wizardSteps: []model.ValidationResponseStep{
 			{
 				Name: "Topics to export",

backend/pkg/connector/patch/iceberg_sink.go

@@ -34,7 +34,7 @@ func NewConfigPatchIcebergSink() *ConfigPatchIcebergSink {
 			Exclude: nil,
 		},
 		ConnectorClassSelector: IncludeExcludeSelector{
-			Include: regexp.MustCompile(`io.tabular.iceberg.connect.IcebergSinkConnector`),
+			Include: regexp.MustCompile(`org.apache.iceberg.connect.IcebergSinkConnector`),
 			Exclude: nil,
 		},
 	}