fix

Author: niksaveliev

ydb/services/persqueue_v1/actors/partition_actor.cpp

@@ -419,6 +419,11 @@ bool FillBatchedData(
         hasOffset = true;
 
         auto proto(GetDeserializedData(r.GetData()));
+
+        if (!proto.has_codec()) {
+            proto.set_codec(NPersQueueCommon::RAW);
+        }
+
         if (proto.GetChunkType() != NKikimrPQClient::TDataChunk::REGULAR) {
             continue; //TODO - no such chunks must be on prod
         }

ydb/services/persqueue_v1/actors/read_session_actor.cpp

@@ -1720,35 +1720,16 @@ void TReadSessionActor<UseMigrationProtocol>::Handle(TEvPQProxy::TEvRead::TPtr&
     ProcessReads(ctx);
 }
 
-
 template <typename TServerMessage>
 i64 TFormedReadResponse<TServerMessage>::ApplyResponse(TServerMessage&& resp) {
     constexpr bool UseMigrationProtocol = std::is_same_v<TServerMessage, PersQueue::V1::MigrationStreamingReadServerMessage>;
 
     if constexpr (UseMigrationProtocol) {
-        auto* partition_data = resp.mutable_data_batch()->mutable_partition_data(0);
-        Y_ABORT_UNLESS(partition_data != nullptr && partition_data->batches_size() > 0);
-
-        for (auto& batch : *partition_data->mutable_batches()) {
-            for (auto& message_data : *batch.mutable_message_data()) {
-                if (message_data.codec() == Ydb::PersQueue::V1::CODEC_UNSPECIFIED) {
-                    message_data.set_codec(Ydb::PersQueue::V1::CODEC_RAW);
-                }
-            }
-        }
-
-        Response.mutable_data_batch()->add_partition_data()->Swap(partition_data);
+        Y_ABORT_UNLESS(resp.data_batch().partition_data_size() == 1);
+        Response.mutable_data_batch()->add_partition_data()->Swap(resp.mutable_data_batch()->mutable_partition_data(0));
     } else {
-        auto* partition_data = resp.mutable_read_response()->mutable_partition_data(0);
-        Y_ABORT_UNLESS(partition_data != nullptr && partition_data->batches_size() > 0);
-
-        for (auto& batch : *partition_data->mutable_batches()) {
-            if (batch.codec() == Ydb::Topic::CODEC_UNSPECIFIED) {
-                batch.set_codec(Ydb::Topic::CODEC_RAW);
-            }
-        }
-
-        Response.mutable_read_response()->add_partition_data()->Swap(partition_data);
+        Y_ABORT_UNLESS(resp.read_response().partition_data_size() == 1);
+        Response.mutable_read_response()->add_partition_data()->Swap(resp.mutable_read_response()->mutable_partition_data(0));
     }
 
     Response.set_status(Ydb::StatusIds::SUCCESS);
@@ -1759,7 +1740,6 @@ i64 TFormedReadResponse<TServerMessage>::ApplyResponse(TServerMessage&& resp) {
 }
 
 
-
 template <typename TServerMessage>
 i64 TFormedReadResponse<TServerMessage>::ApplyDirectReadResponse(TEvPQProxy::TEvDirectReadResponse::TPtr& ev) {
 
@@ -1776,7 +1756,6 @@ i64 TFormedReadResponse<TServerMessage>::ApplyDirectReadResponse(TEvPQProxy::TEv
     return diff;
 }
 
-
 template <bool UseMigrationProtocol>
 void TReadSessionActor<UseMigrationProtocol>::Handle(typename TEvReadResponse::TPtr& ev, const TActorContext& ctx) {
     if (!ActualPartitionActors.contains(ev->Sender)) {