File tree 2 files changed +8
-3
lines changed
topic/src/main/java/tech/ydb/topic/read/impl
2 files changed +8
-3
lines changed Original file line number Diff line number Diff line change @@ -83,8 +83,8 @@ public long getId() {
83
83
return id ;
84
84
}
85
85
86
- public long getFullId () {
87
- return id ;
86
+ public String getFullId () {
87
+ return fullId ;
88
88
}
89
89
90
90
public long getPartitionId () {
@@ -166,7 +166,11 @@ public CompletableFuture<Void> addBatches(List<YdbTopic.StreamReadMessage.ReadRe
166
166
}
167
167
168
168
CompletableFuture .runAsync (() -> decode (newBatch ), decompressionExecutor )
169
- .thenRun (() -> {
169
+ .whenComplete ((res , th ) -> {
170
+ if (th != null ) {
171
+ logger .error ("[{}] Message decoding failed with error: " , fullId , th );
172
+ return ;
173
+ }
170
174
boolean haveNewBatchesReady = false ;
171
175
decodingBatchesLock .lock ();
172
176
Original file line number Diff line number Diff line change @@ -402,6 +402,7 @@ private void onStartPartitionSessionRequest(YdbTopic.StreamReadMessage.StartPart
402
402
.setId (partitionSessionId )
403
403
.setFullId (partitionSessionFullId )
404
404
.setTopicPath (request .getPartitionSession ().getPath ())
405
+ .setConsumerName (consumerName )
405
406
.setPartitionId (partitionId )
406
407
.setCommittedOffset (request .getCommittedOffset ())
407
408
.setPartitionOffsets (new OffsetsRangeImpl (request .getPartitionOffsets ().getStart (),
You can’t perform that action at this time.
0 commit comments