[client] release pending fetch semaphore on discarding

[zuston]

Purpose

Linked issue: close #xxx

Brief change log

Tests

API and Format

Documentation

[fresh-borzoni]

@zuston Nice catch on the permit leak.

Could you add a regression test in LogFetchCollectorTest? CompletedFetch.drain() sets isConsumed=true, so the fix is observable with zero new helpers, I have smth like this in mind:

  @Test                                                                                                                                                                                                                                              
  void testCollectDrainsDiscardedFetch() throws Exception {    
      TableBucket tb = new TableBucket(DATA1_TABLE_ID, 0);                                                                                                                                                                                           
      CompletedFetch completedFetch =                                                                                                                                                                                                                
              makeCompletedFetch(
                      tb,                                                                                                                                                                                                                            
                      new FetchLogResultForBucket(tb,genMemoryLogRecordsByObject(DATA1), 10L), 0L);                                                                                                                                                                                                                           
      logFetchBuffer.add(completedFetch);
      logScannerStatus.unassignScanBuckets(Collections.singletonList(tb));                                                                                                                                                                           
                                                                                                                                                                                                                                                     
      ScanRecords records = logFetchCollector.collectFetch(logFetchBuffer);
                                                                                                                                                                                                                                                     
      assertThat(records.buckets()).isEmpty();                                                                                                                                                                                                       
      assertThat(completedFetch.isConsumed()).isTrue();
  }