Fix FilterUniqueProcessor state leak and endTime filter wrong column

src/batch-doctrine-dbal/src/DoctrineDBALJobExecutionStorage.php

@@ -395,7 +395,7 @@ private function addWheres(Query $query, QueryBuilder $qb): array
         }
 
         if ($query->endTime() !== null) {
-            $qb->andWhere($qb->expr()->isNotNull('start_time'));
+            $qb->andWhere($qb->expr()->isNotNull('end_time'));
         }
         $endDateFrom = $query->endTime()?->getFrom();
         if ($endDateFrom !== null) {

src/batch/src/Job/Item/Processor/FilterUniqueProcessor.php

@@ -6,6 +6,7 @@
 
 use Closure;
 use Yokai\Batch\Job\Item\Exception\SkipItemException;
+use Yokai\Batch\Job\Item\InitializableInterface;
 use Yokai\Batch\Job\Item\ItemProcessorInterface;
 
 /**
@@ -15,7 +16,7 @@
  * it will use a {@see Closure} that will be called for each item
  * and that will be responsible for extracting an identifier from the item.
  */
-final class FilterUniqueProcessor implements ItemProcessorInterface
+final class FilterUniqueProcessor implements ItemProcessorInterface, InitializableInterface
 {
     /**
      * @var array<string, bool>
@@ -60,6 +61,11 @@ public static function withGetter(string $getter): self
         return new self(fn(object $item) => $item->$getter());
     }
 
+    public function initialize(): void
+    {
+        $this->encountered = [];
+    }
+
     public function process(mixed $item): mixed
     {
         $unique = (string)($this->extractUnique)($item);

src/batch/tests/Job/Item/Processor/FilterUniqueProcessorTest.php

@@ -19,16 +19,21 @@ public function test(callable $factory, array $items, array $expected): void
         /** @var FilterUniqueProcessor $processor */
         $processor = $factory();
 
-        $actual = [];
-        foreach ($items as $item) {
-            try {
-                $actual[] = $processor->process($item);
-            } catch (SkipItemException) {
-                //the item have be filtered and not won't be added to $actual
+        // test is done twice to prove initialize method resets state
+        foreach ([1, 2] as $unused) {
+            $processor->initialize();
+
+            $actual = [];
+            foreach ($items as $item) {
+                try {
+                    $actual[] = $processor->process($item);
+                } catch (SkipItemException) {
+                    //the item have be filtered and not won't be added to $actual
+                }
             }
-        }
 
-        self::assertSame($expected, $actual);
+            self::assertSame($expected, $actual);
+        }
     }
 
     public static function provider(): Generator