Merge branch 'v25.3.x' into manual-backport-28935-v25.3.x-95

Author: piyushredpanda

.github/workflows/dispatch-api-docs.yml …thub/workflows/dispatch-docs-updates.yml.github/workflows/dispatch-api-docs.yml renamed to .github/workflows/dispatch-docs-updates.yml .github/workflows/dispatch-api-docs.yml renamed to .github/workflows/dispatch-docs-updates.yml

@@ -1,5 +1,5 @@
 ---
-name: Dispatch Admin API docs update on latest release
+name: Dispatch docs updates on latest release
 
 on:
   release:
@@ -51,8 +51,8 @@ jobs:
             echo "is_latest=false" >> $GITHUB_OUTPUT
           fi
 
-      # Dispatch to docs repo if latest
-      - name: Dispatch to docs repo
+      # Dispatch to api-docs repo if latest
+      - name: Dispatch to api-docs repo
         if: steps.latest.outputs.is_latest == 'true'
         uses: peter-evans/repository-dispatch@v3
         with:
@@ -61,3 +61,14 @@ jobs:
           event-type: redpanda-openapi-bundle
           client-payload: |
             { "tag": "${{ github.event.release.tag_name }}" }
+
+      # Dispatch to docs repo for property docs generation if latest
+      - name: Dispatch property docs generation to docs repo
+        if: steps.latest.outputs.is_latest == 'true'
+        uses: peter-evans/repository-dispatch@v3
+        with:
+          token: ${{ env.ACTIONS_BOT_TOKEN }}
+          repository: redpanda-data/docs
+          event-type: trigger-property-docs-generation
+          client-payload: |
+            { "tag": "${{ github.event.release.tag_name }}" }

MODULE.bazel.lock

@@ -40,9 +40,9 @@ def data_dependency():
     http_archive(
         name = "c-ares",
         build_file = "//bazel/thirdparty:c-ares.BUILD",
-        sha256 = "7d935790e9af081c25c495fd13c2cfcda4792983418e96358ef6e7320ee06346",
-        strip_prefix = "c-ares-1.34.5",
-        url = "https://vectorized-public.s3.amazonaws.com/dependencies/c-ares-1.34.5.tar.gz",
+        sha256 = "912dd7cc3b3e8a79c52fd7fb9c0f4ecf0aaa73e45efda880266a2d6e26b84ef5",
+        strip_prefix = "c-ares-1.34.6",
+        url = "https://vectorized-public.s3.amazonaws.com/dependencies/c-ares-1.34.6.tar.gz",
     )
 
     http_archive(

bazel/repositories.bzl

@@ -34,6 +34,7 @@ const (
 	secTopicSync      = "Topic Sync"
 	secConsumerOffset = "Consumer Offset Sync"
 	secSecurity       = "Security Sync"
+	secSchemaRegistry = "Schema Registry Sync"
 )
 
 func newDescribeCommand(fs afero.Fs, p *config.Params) *cobra.Command {
@@ -109,6 +110,7 @@ Display only the client configuration:
 	cmd.Flags().BoolVarP(&opts.topic, "print-topic", "t", false, "Print the detailed topic configuration section")
 	cmd.Flags().BoolVarP(&opts.co, "print-consumer", "r", false, "Print the detailed consumer offset configuration section")
 	cmd.Flags().BoolVarP(&opts.sec, "print-security", "s", false, "Print the detailed security configuration section")
+	cmd.Flags().BoolVarP(&opts.sr, "print-registry", "y", false, "Print the detailed schema registry configuration section")
 	cmd.Flags().BoolVarP(&opts.all, "print-all", "a", false, "Print all sections")
 	return cmd
 }
@@ -120,16 +122,17 @@ type slDescribeOptions struct {
 	topic    bool
 	co       bool // consumer offset
 	sec      bool // security
+	sr       bool // schema registry
 }
 
 // If no flags are set, default to overview and client sections.
 func (o *slDescribeOptions) defaultOrAll() {
-	if !o.all && !o.overview && !o.client && !o.topic && !o.co && !o.sec {
+	if !o.all && !o.overview && !o.client && !o.topic && !o.co && !o.sec && !o.sr {
 		o.overview, o.client = true, true
 	}
 
 	if o.all {
-		o.overview, o.client, o.topic, o.co, o.sec = true, true, true, true, true
+		o.overview, o.client, o.topic, o.co, o.sec, o.sr = true, true, true, true, true, true
 	}
 }
 
@@ -141,6 +144,7 @@ func printShadowLinkDescription(link *adminv2.ShadowLink, opts slDescribeOptions
 			secTopicSync:      opts.topic,
 			secConsumerOffset: opts.co,
 			secSecurity:       opts.sec,
+			secSchemaRegistry: opts.sr,
 		})...,
 	)
 
@@ -165,6 +169,10 @@ func printShadowLinkDescription(link *adminv2.ShadowLink, opts slDescribeOptions
 	sections.Add(secSecurity, func() {
 		printSecuritySync(cfg.GetSecuritySyncOptions())
 	})
+
+	sections.Add(secSchemaRegistry, func() {
+		printSchemaRegistrySync(cfg.GetSchemaRegistrySyncOptions())
+	})
 }
 
 func printCloudShadowLinkDescription(link *controlplanev1.ShadowLink, opts slDescribeOptions) {
@@ -197,6 +205,10 @@ func printCloudShadowLinkDescription(link *controlplanev1.ShadowLink, opts slDes
 	sections.Add(secSecurity, func() {
 		printSecuritySync(link.GetSecuritySyncOptions())
 	})
+
+	sections.Add(secSchemaRegistry, func() {
+		printSchemaRegistrySync(link.GetSchemaRegistrySyncOptions())
+	})
 }
 
 func printOverview(link *adminv2.ShadowLink) {
@@ -448,6 +460,16 @@ func printSecuritySync(opts *adminv2.SecuritySettingsSyncOptions) {
 	}
 }
 
+func printSchemaRegistrySync(opts *adminv2.SchemaRegistrySyncOptions) {
+	tw := out.NewTabWriter()
+	defer tw.Flush()
+	if opts == nil {
+		tw.Print("No schema registry sync configuration")
+		return
+	}
+	tw.Print("SHADOWING MODE", strings.ReplaceAll(opts.WhichSchemaRegistryShadowingMode().String(), "_", " "))
+}
+
 func formatScramMechanism(m adminv2.ScramMechanism) string {
 	switch m {
 	case adminv2.ScramMechanism_SCRAM_MECHANISM_SCRAM_SHA_256:

src/go/rpk/pkg/cli/shadow/describe.go

@@ -682,8 +682,14 @@ func (c *consumer) filterEmptyPartitions() (allEmpty bool) {
 			for lp, lat := range lps {
 				rat, exists := rps[lp]
 				if exists && rat <= lat {
+					zap.L().Sugar().Warnf("no data to consume for %s/%d: requested end offset %d, available start offset %d",
+						lt, lp, rat, lat)
 					delete(rps, lp)
 					delete(lps, lp)
+				} else if rat > lat {
+					// There is data to consume. We consume from the available start offset (lat),
+					// which automatically adjusts if the user's requested start was too low.
+					zap.L().Sugar().Debugf("consuming %s/%d from offset %d", lt, lp, lat)
 				}
 			}
 		}

src/go/rpk/pkg/cli/topic/consume.go

@@ -128,6 +128,22 @@ struct archival_metadata_stm_base_fixture
           .get();
     }
 
+    void populate_log_with_data(size_t record_count) {
+        for (size_t i = 0; i < record_count; ++i) {
+            auto batch = model::test::make_random_batch(
+              model::test::record_batch_spec{
+                .count = 1,
+                .bt = model::record_batch_type::raft_data,
+              });
+            batch.header().ctx.term = model::term_id(1);
+            auto appender = _raft->log()->make_appender(
+              storage::log_append_config{});
+            auto rdr = model::make_memory_record_batch_reader(std::move(batch));
+            rdr.for_each_ref(std::move(appender), model::no_timeout).get();
+        }
+        _raft->log()->flush().get();
+    }
+
     ~archival_metadata_stm_base_fixture() override {
         stop_all();
         cloud_conn_pool.local().shutdown_connections();
@@ -319,6 +335,7 @@ void check_snapshot_size(
 
 FIXTURE_TEST(test_snapshot_loading, archival_metadata_stm_base_fixture) {
     create_raft();
+    populate_log_with_data(43);
     auto& ntp_cfg = _raft->log_config();
     partition_manifest m(ntp_cfg.ntp(), ntp_cfg.get_remote_revision());
     m.add(
@@ -400,7 +417,9 @@ FIXTURE_TEST(test_snapshot_loading, archival_metadata_stm_base_fixture) {
     }
 
     BOOST_REQUIRE_EQUAL(archival_stm->get_start_offset(), model::offset{100});
-    BOOST_REQUIRE(archival_stm->manifest() == m);
+    BOOST_REQUIRE(
+      archival_stm->manifest().make_manifest_metadata()
+      == m.make_manifest_metadata());
     check_snapshot_size(*archival_stm, ntp_cfg);
 
     // A snapshot constructed with make_snapshot is always clean
@@ -411,6 +430,7 @@ FIXTURE_TEST(test_snapshot_loading, archival_metadata_stm_base_fixture) {
 
 FIXTURE_TEST(test_sname_derivation, archival_metadata_stm_base_fixture) {
     create_raft();
+    populate_log_with_data(50);
     auto& ntp_cfg = _raft->log_config();
     partition_manifest m(ntp_cfg.ntp(), ntp_cfg.get_remote_revision());
 
@@ -670,57 +690,6 @@ ss::future<> make_old_snapshot(
       tmp_snapshot_mgr, std::move(snapshot));
 }
 
-FIXTURE_TEST(
-  test_archival_metadata_stm_snapshot_version_compatibility,
-  archival_metadata_stm_base_fixture) {
-    create_raft();
-    auto& ntp_cfg = _raft->log_config();
-    partition_manifest m(ntp_cfg.ntp(), ntp_cfg.get_remote_revision());
-    m.add(
-      segment_name("0-1-v1.log"),
-      segment_meta{
-        .base_offset = model::offset(0),
-        .committed_offset = model::offset(99),
-        .archiver_term = model::term_id(1),
-        .segment_term = model::term_id(1),
-      });
-    m.add(
-      segment_name("100-1-v1.log"),
-      segment_meta{
-        .base_offset = model::offset(100),
-        .committed_offset = model::offset(199),
-        .archiver_term = model::term_id(1),
-        .segment_term = model::term_id(1),
-      });
-    m.add(
-      segment_name("200-1-v1.log"),
-      segment_meta{
-        .base_offset = model::offset(200),
-        .committed_offset = model::offset(299),
-        .archiver_term = model::term_id(1),
-        .segment_term = model::term_id(1),
-      });
-    m.advance_insync_offset(model::offset(3));
-
-    make_old_snapshot(ntp_cfg, m, model::offset{3}).get();
-
-    raft::state_machine_manager_builder builder;
-    auto archival_stm = builder.create_stm<cluster::archival_metadata_stm>(
-      _raft.get(),
-      cloud_api.local(),
-      _feature_table.local(),
-      logger,
-      std::nullopt,
-      std::nullopt);
-
-    _raft->start(std::move(builder)).get();
-    _started = true;
-    wait_for_confirmed_leader();
-
-    BOOST_REQUIRE(archival_stm->manifest() == m);
-    check_snapshot_size(*archival_stm, ntp_cfg);
-}
-
 FIXTURE_TEST(test_archival_stm_batching, archival_metadata_stm_fixture) {
     wait_for_confirmed_leader();
     std::vector<cloud_storage::segment_meta> m;

src/v/cluster/archival/tests/archival_metadata_stm_test.cc

@@ -1089,7 +1089,7 @@ errc frontend::validator::validate_metadata_mirroring_config(
 }
 
 ss::future<errc> frontend::failover_link_topics(
-  ::cluster_link::model::id_t id, model::timeout_clock::time_point timeout) {
+  ::cluster_link::model::id_t id, model::timeout_clock::duration timeout) {
     auto meta = _table->find_link_by_id(id);
     if (!meta.has_value()) {
         co_return errc::does_not_exist;
@@ -1136,7 +1136,7 @@ ss::future<errc> frontend::failover_link_topics(
                    {.topic = t,
                     .status
                     = ::cluster_link::model::mirror_topic_status::failing_over},
-                   timeout)
+                   model::timeout_clock::now() + timeout)
             .then([&errors](errc err_code) {
                 if (err_code != errc::success) {
                     errors.push_back(err_code);

src/v/cluster/cluster_link/frontend.cc

@@ -78,7 +78,7 @@ class frontend : public ss::peering_sharded_service<frontend> {
       model::timeout_clock::time_point);
 
     ss::future<errc> failover_link_topics(
-      ::cluster_link::model::id_t, model::timeout_clock::time_point);
+      ::cluster_link::model::id_t, model::timeout_clock::duration);
 
     bool cluster_link_active() const;
 

src/v/cluster/cluster_link/frontend.h

@@ -682,6 +682,7 @@ ss::future<> controller::start(
       std::ref(_members_table),
       std::ref(_partition_balancer),
       std::ref(_partition_manager),
+      std::ref(_partition_leaders),
       std::ref(_as));
 
     co_await set_raft_manager_remake_cb();