segmentstore/server/src/test/java/io/pravega/segmentstore/server/tables/WriterTableProcessorTests.java
/**
     * Tests the ability to reconcile the {@link TableAttributes#INDEX_OFFSET} value when that changes behind the scenes.
     */
Tests the ability to reconcile the  TableAttributes#INDEX_OFFSET value when that changes behind the scenes.
Tests the ability to reconcile the TableAttributes#INDEX_OFFSET value when that changes behind the scenes .
@Test
    public void testReconcileTableIndexOffset() throws Exception {
        @Cleanup
        val context = new TestContext();
        val e1 = TableEntry.unversioned(new ByteArraySegment("k1".getBytes()), new ByteArraySegment("v1".getBytes()));
        val e2 = TableEntry.unversioned(new ByteArraySegment("k2".getBytes()), new ByteArraySegment("v2".getBytes()));
        val append1 = generateRawAppend(e1, INITIAL_LAST_INDEXED_OFFSET, context);
        val append2 = generateRawAppend(e2, append1.getLastStreamSegmentOffset(), context);
        context.segmentMock.append(append1.getData(), null, TIMEOUT).join();
        context.segmentMock.append(append2.getData(), null, TIMEOUT).join();
        context.processor.add(new CachedStreamSegmentAppendOperation(append1));
        context.processor.add(new CachedStreamSegmentAppendOperation(append2));
        context.metadata.updateAttributes(Collections.singletonMap(TableAttributes.INDEX_OFFSET, INITIAL_LAST_INDEXED_OFFSET - 1));
        int attributeCountBefore = context.segmentMock.getAttributeCount();
        AssertExtensions.assertSuppliedFutureThrows(
                "flush() worked when INDEX_OFFSET decreased.",
                () -> context.processor.flush(TIMEOUT),
                ex -> ex instanceof DataCorruptionException);
        int attributeCountAfter = context.segmentMock.getAttributeCount();
        Assert.assertEquals("flush() seems to have modified the index after failed attempt", attributeCountBefore, attributeCountAfter);
        Assert.assertEquals("flush() seems to have modified the index after failed attempt.",
                INITIAL_LAST_INDEXED_OFFSET - 1, context.indexReader.getLastIndexedOffset(context.metadata));
        context.metadata.updateAttributes(Collections.singletonMap(TableAttributes.INDEX_OFFSET, INITIAL_LAST_INDEXED_OFFSET + 1));
        attributeCountBefore = context.segmentMock.getAttributeCount();
        AssertExtensions.assertSuppliedFutureThrows(
                "flush() worked when INDEX_OFFSET changed to middle of append.",
                () -> context.processor.flush(TIMEOUT),
                ex -> ex instanceof DataCorruptionException);
        attributeCountAfter = context.segmentMock.getAttributeCount();
        Assert.assertEquals("flush() seems to have modified the index after failed attempt", attributeCountBefore, attributeCountAfter);
        Assert.assertEquals("flush() seems to have modified the index after failed attempt.",
                INITIAL_LAST_INDEXED_OFFSET + 1, context.indexReader.getLastIndexedOffset(context.metadata));
        context.metadata.updateAttributes(Collections.singletonMap(TableAttributes.INDEX_OFFSET, append2.getStreamSegmentOffset()));
        attributeCountBefore = context.segmentMock.getAttributeCount();
        context.processor.flush(TIMEOUT).join();
        attributeCountAfter = context.segmentMock.getAttributeCount();
        AssertExtensions.assertGreaterThan("flush() did not modify the index partial reconciliation.", attributeCountBefore, attributeCountAfter);
        Assert.assertEquals("flush() did not modify the index partial reconciliation.",
                append2.getLastStreamSegmentOffset(), context.indexReader.getLastIndexedOffset(context.metadata));
        Assert.assertFalse("Unexpected result from mustFlush() after partial reconciliation.", context.processor.mustFlush());
        val e3 = TableEntry.unversioned(new ByteArraySegment("k3".getBytes()), new ByteArraySegment("v3".getBytes()));
        val append3 = generateRawAppend(e3, append2.getLastStreamSegmentOffset(), context);
        context.segmentMock.append(append3.getData(), null, TIMEOUT).join();
        context.processor.add(new CachedStreamSegmentAppendOperation(append3));
        context.metadata.updateAttributes(Collections.singletonMap(TableAttributes.INDEX_OFFSET, append3.getLastStreamSegmentOffset() + 1));
        attributeCountBefore = context.segmentMock.getAttributeCount();
        context.processor.flush(TIMEOUT).join();
        attributeCountAfter = context.segmentMock.getAttributeCount();
        Assert.assertEquals("flush() seems to have modified the index after full reconciliation.", attributeCountBefore, attributeCountAfter);
        Assert.assertEquals("flush() did not properly update INDEX_OFFSET after full reconciliation.",
                append3.getLastStreamSegmentOffset() + 1, context.indexReader.getLastIndexedOffset(context.metadata));
        Assert.assertFalse("Unexpected result from mustFlush() after full reconciliation.", context.processor.mustFlush());
    }
@ Test public void testReconcileTableIndexOffset ( ) throws Exception { @ Cleanup val context = new TestContext ( ) ; val e1 = TableEntry . unversioned ( new ByteArraySegment ( "k1" . getBytes ( ) ) , new ByteArraySegment ( "v1" . getBytes ( ) ) ) ; val e2 = TableEntry . unversioned ( new ByteArraySegment ( "k2" . getBytes ( ) ) , new ByteArraySegment ( "v2" . getBytes ( ) ) ) ; val append1 = generateRawAppend ( e1 , INITIAL_LAST_INDEXED_OFFSET , context ) ; val append2 = generateRawAppend ( e2 , append1 . getLastStreamSegmentOffset ( ) , context ) ; context . segmentMock . append ( append1 . getData ( ) , null , TIMEOUT ) . join ( ) ; context . segmentMock . append ( append2 . getData ( ) , null , TIMEOUT ) . join ( ) ; context . processor . add ( new CachedStreamSegmentAppendOperation ( append1 ) ) ; context . processor . add ( new CachedStreamSegmentAppendOperation ( append2 ) ) ; context . metadata . updateAttributes ( Collections . singletonMap ( TableAttributes . INDEX_OFFSET , INITIAL_LAST_INDEXED_OFFSET - 1 ) ) ; int attributeCountBefore = context . segmentMock . getAttributeCount ( ) ; AssertExtensions . assertSuppliedFutureThrows ( "flush() worked when INDEX_OFFSET decreased." , ( ) -> context . processor . flush ( TIMEOUT ) , ex -> ex instanceof DataCorruptionException ) ; int attributeCountAfter = context . segmentMock . getAttributeCount ( ) ; Assert . assertEquals ( "flush() seems to have modified the index after failed attempt" , attributeCountBefore , attributeCountAfter ) ; Assert . assertEquals ( "flush() seems to have modified the index after failed attempt." , INITIAL_LAST_INDEXED_OFFSET - 1 , context . indexReader . getLastIndexedOffset ( context . metadata ) ) ; context . metadata . updateAttributes ( Collections . singletonMap ( TableAttributes . INDEX_OFFSET , INITIAL_LAST_INDEXED_OFFSET + 1 ) ) ; attributeCountBefore = context . segmentMock . getAttributeCount ( ) ; AssertExtensions . assertSuppliedFutureThrows ( "flush() worked when INDEX_OFFSET changed to middle of append." , ( ) -> context . processor . flush ( TIMEOUT ) , ex -> ex instanceof DataCorruptionException ) ; attributeCountAfter = context . segmentMock . getAttributeCount ( ) ; Assert . assertEquals ( "flush() seems to have modified the index after failed attempt" , attributeCountBefore , attributeCountAfter ) ; Assert . assertEquals ( "flush() seems to have modified the index after failed attempt." , INITIAL_LAST_INDEXED_OFFSET + 1 , context . indexReader . getLastIndexedOffset ( context . metadata ) ) ; context . metadata . updateAttributes ( Collections . singletonMap ( TableAttributes . INDEX_OFFSET , append2 . getStreamSegmentOffset ( ) ) ) ; attributeCountBefore = context . segmentMock . getAttributeCount ( ) ; context . processor . flush ( TIMEOUT ) . join ( ) ; attributeCountAfter = context . segmentMock . getAttributeCount ( ) ; AssertExtensions . assertGreaterThan ( "flush() did not modify the index partial reconciliation." , attributeCountBefore , attributeCountAfter ) ; Assert . assertEquals ( "flush() did not modify the index partial reconciliation." , append2 . getLastStreamSegmentOffset ( ) , context . indexReader . getLastIndexedOffset ( context . metadata ) ) ; Assert . assertFalse ( "Unexpected result from mustFlush() after partial reconciliation." , context . processor . mustFlush ( ) ) ; val e3 = TableEntry . unversioned ( new ByteArraySegment ( "k3" . getBytes ( ) ) , new ByteArraySegment ( "v3" . getBytes ( ) ) ) ; val append3 = generateRawAppend ( e3 , append2 . getLastStreamSegmentOffset ( ) , context ) ; context . segmentMock . append ( append3 . getData ( ) , null , TIMEOUT ) . join ( ) ; context . processor . add ( new CachedStreamSegmentAppendOperation ( append3 ) ) ; context . metadata . updateAttributes ( Collections . singletonMap ( TableAttributes . INDEX_OFFSET , append3 . getLastStreamSegmentOffset ( ) + 1 ) ) ; attributeCountBefore = context . segmentMock . getAttributeCount ( ) ; context . processor . flush ( TIMEOUT ) . join ( ) ; attributeCountAfter = context . segmentMock . getAttributeCount ( ) ; Assert . assertEquals ( "flush() seems to have modified the index after full reconciliation." , attributeCountBefore , attributeCountAfter ) ; Assert . assertEquals ( "flush() did not properly update INDEX_OFFSET after full reconciliation." , append3 . getLastStreamSegmentOffset ( ) + 1 , context . indexReader . getLastIndexedOffset ( context . metadata ) ) ; Assert . assertFalse ( "Unexpected result from mustFlush() after full reconciliation." , context . processor . mustFlush ( ) ) ; }
Tests the ability to reconcile the {@link TableAttributes#INDEX_OFFSET} value when that changes behind the scenes.
Tests the ability to reconcile the { @link TableAttributes#INDEX_OFFSET } value when that changes behind the scenes .
// Generate two TableEntries, write them to the segment and queue them into the processor. // 1. INDEX_OFFSET changes to smaller than first append // 2. INDEX_OFFSET changes to middle of append. // 3. INDEX_OFFSET changes after the first append, but before the second one. // 4. INDEX_OFFSET changes beyond the last append.