fix

Author: niksaveliev

ydb/core/persqueue/dread_cache_service/caching_service.cpp

@@ -475,6 +475,10 @@ class TPQDirectReadCacheService : public TActorBootstrapped<TPQDirectReadCacheSe
                 continue; //TODO - no such chunks must be on prod
             }
 
+            if (!proto.has_codec()) {
+                proto.set_codec(NPersQueueCommon::RAW);
+            }
+
             TString sourceId;
             if (!r.GetSourceId().empty()) {
                 sourceId = NPQ::NSourceIdEncoding::Decode(r.GetSourceId());

ydb/services/datastreams/datastreams_proxy.cpp

@@ -1664,9 +1664,12 @@ namespace NKikimr::NDataStreams::V1 {
                 record->set_approximate_arrival_timestamp(r.GetCreateTimestampMS());
                 record->set_partition_key(r.GetPartitionKey());
                 record->set_sequence_number(std::to_string(r.GetOffset()).c_str());
-                if (proto.GetCodec() > 0) {
-                    record->set_codec(proto.GetCodec() + 1);
+
+                if (!proto.has_codec()) {
+                    proto.set_codec(NPersQueueCommon::RAW);
                 }
+
+                record->set_codec(proto.GetCodec() + 1);
             }
             if (!results.empty()) {
                 auto last = results.rbegin();

ydb/services/deprecated/persqueue_v0/grpc_pq_read_actor.cpp

@@ -2137,6 +2137,11 @@ void TPartitionActor::Handle(TEvPersQueue::TEvResponse::TPtr& ev, const TActorCo
         if (proto.GetChunkType() != NKikimrPQClient::TDataChunk::REGULAR) {
             continue; //TODO - no such chunks must be on prod
         }
+
+        if (!proto.has_codec()) {
+            proto.set_codec(NPersQueueCommon::RAW);
+        }
+
         TString sourceId = "";
         if (!r.GetSourceId().empty()) {
             if (!NPQ::NSourceIdEncoding::IsValidEncoded(r.GetSourceId())) {