Fix concurrent append to interval with only unused segments (#18216)

Bug:
Concurrent append uses lock of type APPEND which always uses a lock version of epoch 1970-01-01.

This can cause data loss in a flow as follows:
- Ingest data using an APPEND task to an empty interval
- Mark all the segments as unused
- Re-run the APPEND task
- Data is not visible since old segment IDs (now unused) are allocated again

Fix:
In segment allocation, do not reuse an old segment ID, used or unused.
This fix was already done for some cases back in #16380 .
An embedded test for this has been included in #18207

Author: kfaraz

indexing-service/src/test/java/org/apache/druid/indexing/common/task/concurrent/ConcurrentReplaceAndAppendTest.java

@@ -1119,6 +1119,38 @@ public void testLockAllocateDayReplaceMonthAllocateAppend()
     verifyIntervalHasVisibleSegments(JAN_23, segmentV10, segmentV11, segmentV12);
   }
 
+  @Test
+  public void test_concurrentAppend_toIntervalWithUnusedSegments()
+  {
+    // Allocate and commit an APPEND segment
+    final SegmentIdWithShardSpec pendingSegment
+        = appendTask.allocateSegmentForTimestamp(FIRST_OF_JAN_23.getStart(), Granularities.DAY);
+    Assert.assertEquals(SEGMENT_V0, pendingSegment.getVersion());
+    Assert.assertEquals(0, pendingSegment.getShardSpec().getPartitionNum());
+
+    final DataSegment segmentV01 = asSegment(pendingSegment);
+    appendTask.commitAppendSegments(segmentV01);
+
+    verifyIntervalHasUsedSegments(FIRST_OF_JAN_23, segmentV01);
+    verifyIntervalHasVisibleSegments(FIRST_OF_JAN_23, segmentV01);
+
+    // Mark it as unused
+    getStorageCoordinator().markAllSegmentsAsUnused(appendTask.getDataSource());
+    verifyIntervalHasUsedSegments(FIRST_OF_JAN_23);
+
+    // Allocate and commit another APPEND segment
+    final SegmentIdWithShardSpec pendingSegment2
+        = appendTask.allocateSegmentForTimestamp(FIRST_OF_JAN_23.getStart(), Granularities.DAY);
+    Assert.assertEquals(SEGMENT_V0, pendingSegment2.getVersion());
+    Assert.assertEquals(1, pendingSegment2.getShardSpec().getPartitionNum());
+
+    final DataSegment segmentV02 = asSegment(pendingSegment2);
+    appendTask.commitAppendSegments(segmentV02);
+    Assert.assertNotEquals(segmentV01, segmentV02);
+
+    verifyIntervalHasUsedSegments(FIRST_OF_JAN_23, segmentV02);
+    verifyIntervalHasVisibleSegments(FIRST_OF_JAN_23, segmentV02);
+  }
 
   @Nullable
   private DataSegment findSegmentWith(String version, Map<String, Object> loadSpec, Set<DataSegment> segments)

server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java

@@ -1419,6 +1419,7 @@ private PendingSegmentRecord createNewPendingSegment(
           version,
           partialShardSpec.complete(jsonMapper, newPartitionId, 0)
       );
+      pendingSegmentId = getTrueAllocatedId(transaction, pendingSegmentId);
       return PendingSegmentRecord.create(
           pendingSegmentId,
           request.getSequenceName(),
@@ -1555,12 +1556,13 @@ private SegmentIdWithShardSpec createNewPendingSegment(
                                  ? PartitionIds.NON_ROOT_GEN_START_PARTITION_ID
                                  : PartitionIds.ROOT_GEN_START_PARTITION_ID;
       String version = newSegmentVersion == null ? existingVersion : newSegmentVersion;
-      return new SegmentIdWithShardSpec(
+      SegmentIdWithShardSpec allocatedId = new SegmentIdWithShardSpec(
           dataSource,
           interval,
           version,
           partialShardSpec.complete(jsonMapper, newPartitionId, 0)
       );
+      return getTrueAllocatedId(transaction, allocatedId);
     } else if (!overallMaxId.getInterval().equals(interval)) {
       log.warn(
           "Cannot allocate new segment for dataSource[%s], interval[%s], existingVersion[%s]: conflicting segment[%s].",