@@ -315,8 +315,9 @@ void testSyncMetadataTable() throws Exception {
315315 reset ();
316316 // override the default configuration
317317 Configuration conf = TestConfigurations .getDefaultConf (tempFile .getAbsolutePath ());
318+ int metadataCompactionDeltaCommits = 5 ;
318319 conf .setBoolean (FlinkOptions .METADATA_ENABLED , true );
319- conf .setInteger (FlinkOptions .METADATA_COMPACTION_DELTA_COMMITS , 5 );
320+ conf .setInteger (FlinkOptions .METADATA_COMPACTION_DELTA_COMMITS , metadataCompactionDeltaCommits );
320321 OperatorCoordinator .Context context = new MockOperatorCoordinatorContext (new OperatorID (), 1 );
321322 coordinator = new StreamWriteOperatorCoordinator (conf , context );
322323 coordinator .start ();
@@ -332,23 +333,27 @@ void testSyncMetadataTable() throws Exception {
332333 final String metadataTableBasePath = HoodieTableMetadata .getMetadataTableBasePath (tempFile .getAbsolutePath ());
333334 HoodieTableMetaClient metadataTableMetaClient = HoodieTestUtils .createMetaClient (new HadoopStorageConfiguration (HadoopConfigurations .getHadoopConf (conf )), metadataTableBasePath );
334335 HoodieTimeline completedTimeline = metadataTableMetaClient .getActiveTimeline ().filterCompletedInstants ();
335- assertThat ("One instant need to sync to metadata table" , completedTimeline .countInstants (), is (1 ));
336+ HoodieTableMetaClient dataTableMetaClient =
337+ HoodieTestUtils .createMetaClient (new HadoopStorageConfiguration (HadoopConfigurations .getHadoopConf (conf )), new Path (metadataTableBasePath ).getParent ().getParent ().toString ());
338+ int metadataPartitions = dataTableMetaClient .getTableConfig ().getMetadataPartitions ().size ();
339+ assertThat ("Instants needed to sync to metadata table do not match" , completedTimeline .countInstants (), is (metadataPartitions ));
336340 assertThat (completedTimeline .lastInstant ().get ().getTimestamp (), startsWith (HoodieTableMetadata .SOLO_COMMIT_TIMESTAMP ));
337341
338342 // test metadata table compaction
339- // write another 4 commits
340- for (int i = 1 ; i < 5 ; i ++) {
343+ // write few more commits until compaction
344+ int numCommits ;
345+ for (numCommits = metadataPartitions ; numCommits < metadataCompactionDeltaCommits ; numCommits ++) {
341346 instant = mockWriteWithMetadata ();
342347 metadataTableMetaClient .reloadActiveTimeline ();
343348 completedTimeline = metadataTableMetaClient .getActiveTimeline ().filterCompletedInstants ();
344- assertThat ("One instant need to sync to metadata table" , completedTimeline .countInstants (), is (i + 1 ));
349+ assertThat ("One instant need to sync to metadata table" , completedTimeline .countInstants (), is (numCommits + 1 ));
345350 assertThat (completedTimeline .lastInstant ().get ().getTimestamp (), is (instant ));
346351 }
347352 // the 5th commit triggers the compaction
348353 mockWriteWithMetadata ();
349354 metadataTableMetaClient .reloadActiveTimeline ();
350355 completedTimeline = metadataTableMetaClient .reloadActiveTimeline ().filterCompletedAndCompactionInstants ();
351- assertThat ("One instant need to sync to metadata table" , completedTimeline .countInstants (), is (7 ));
356+ assertThat ("One instant need to sync to metadata table" , completedTimeline .countInstants (), is (numCommits + 2 ));
352357 assertThat (completedTimeline .nthFromLastInstant (0 ).get ().getAction (), is (HoodieTimeline .COMMIT_ACTION ));
353358 // write another 2 commits
354359 for (int i = 7 ; i < 8 ; i ++) {
@@ -401,24 +406,28 @@ void testSyncMetadataTableWithLogCompaction() throws Exception {
401406 final String metadataTableBasePath = HoodieTableMetadata .getMetadataTableBasePath (tempFile .getAbsolutePath ());
402407 HoodieTableMetaClient metadataTableMetaClient = HoodieTestUtils .createMetaClient (new HadoopStorageConfiguration (HadoopConfigurations .getHadoopConf (conf )), metadataTableBasePath );
403408 HoodieTimeline completedTimeline = metadataTableMetaClient .getActiveTimeline ().filterCompletedInstants ();
404- assertThat ("One instant need to sync to metadata table" , completedTimeline .countInstants (), is (1 ));
409+ HoodieTableMetaClient dataTableMetaClient =
410+ HoodieTestUtils .createMetaClient (new HadoopStorageConfiguration (HadoopConfigurations .getHadoopConf (conf )), new Path (metadataTableBasePath ).getParent ().getParent ().toString ());
411+ int metadataPartitions = dataTableMetaClient .getTableConfig ().getMetadataPartitions ().size ();
412+ assertThat ("Instants needed to sync to metadata table do not match" , completedTimeline .countInstants (), is (metadataPartitions ));
405413 assertThat (completedTimeline .lastInstant ().get ().getTimestamp (), startsWith (HoodieTableMetadata .SOLO_COMMIT_TIMESTAMP ));
406414
407415 // test metadata table log compaction
408416 // already 1 commit is used to initialized FILES partition in MDT
409417 // write another 4 commits
410- for (int i = 1 ; i < 5 ; i ++) {
418+ int numCommits ;
419+ for (numCommits = metadataPartitions ; numCommits < metadataPartitions + 4 ; numCommits ++) {
411420 instant = mockWriteWithMetadata ();
412421 metadataTableMetaClient .reloadActiveTimeline ();
413422 completedTimeline = metadataTableMetaClient .getActiveTimeline ().filterCompletedInstants ();
414- assertThat ("One instant need to sync to metadata table" , completedTimeline .countInstants (), is (i + 1 ));
423+ assertThat ("One instant need to sync to metadata table" , completedTimeline .countInstants (), is (numCommits + 1 ));
415424 assertThat (completedTimeline .lastInstant ().get ().getTimestamp (), is (instant ));
416425 }
417426 // the 5th commit triggers the log compaction
418427 mockWriteWithMetadata ();
419428 metadataTableMetaClient .reloadActiveTimeline ();
420429 completedTimeline = metadataTableMetaClient .reloadActiveTimeline ().filterCompletedAndCompactionInstants ();
421- assertThat ("One instant need to sync to metadata table" , completedTimeline .countInstants (), is (7 ));
430+ assertThat ("One instant need to sync to metadata table" , completedTimeline .countInstants (), is (numCommits + 2 ));
422431 assertThat ("The log compaction instant time should be new generated" ,
423432 completedTimeline .nthFromLastInstant (1 ).get ().getTimestamp (), not (instant ));
424433 // log compaction is another delta commit
@@ -447,7 +456,10 @@ void testSyncMetadataTableWithRollback() throws Exception {
447456 final String metadataTableBasePath = HoodieTableMetadata .getMetadataTableBasePath (tempFile .getAbsolutePath ());
448457 HoodieTableMetaClient metadataTableMetaClient = HoodieTestUtils .createMetaClient (new HadoopStorageConfiguration (HadoopConfigurations .getHadoopConf (conf )), metadataTableBasePath );
449458 HoodieTimeline completedTimeline = metadataTableMetaClient .getActiveTimeline ().filterCompletedInstants ();
450- assertThat ("One instant need to sync to metadata table" , completedTimeline .countInstants (), is (1 ));
459+ HoodieTableMetaClient dataTableMetaClient =
460+ HoodieTestUtils .createMetaClient (new HadoopStorageConfiguration (HadoopConfigurations .getHadoopConf (conf )), new Path (metadataTableBasePath ).getParent ().getParent ().toString ());
461+ int metadataPartitions = dataTableMetaClient .getTableConfig ().getMetadataPartitions ().size ();
462+ assertThat ("Instants needed to sync to metadata table do not match" , completedTimeline .countInstants (), is (metadataPartitions ));
451463 assertThat (completedTimeline .lastInstant ().get ().getTimestamp (), startsWith (HoodieTableMetadata .SOLO_COMMIT_TIMESTAMP ));
452464
453465 // writes a normal commit
@@ -464,7 +476,7 @@ void testSyncMetadataTableWithRollback() throws Exception {
464476 metadataTableMetaClient .reloadActiveTimeline ();
465477
466478 completedTimeline = metadataTableMetaClient .getActiveTimeline ().filterCompletedInstants ();
467- assertThat ("One instant need to sync to metadata table" , completedTimeline .countInstants (), is (4 ));
479+ assertThat ("One instant need to sync to metadata table" , completedTimeline .countInstants (), is (metadataPartitions + 3 ));
468480 assertThat (completedTimeline .nthFromLastInstant (1 ).get ().getTimestamp (), is (instant ));
469481 assertThat ("The pending instant should be rolled back first" ,
470482 completedTimeline .lastInstant ().get ().getAction (), is (HoodieTimeline .ROLLBACK_ACTION ));
@@ -530,13 +542,16 @@ void testLockForMetadataTable() throws Exception {
530542 final String metadataTableBasePath = HoodieTableMetadata .getMetadataTableBasePath (tempFile .getAbsolutePath ());
531543 HoodieTableMetaClient metadataTableMetaClient = HoodieTestUtils .createMetaClient (new HadoopStorageConfiguration (HadoopConfigurations .getHadoopConf (conf )), metadataTableBasePath );
532544 HoodieTimeline completedTimeline = metadataTableMetaClient .getActiveTimeline ().filterCompletedInstants ();
533- assertThat ("One instant need to sync to metadata table" , completedTimeline .countInstants (), is (1 ));
545+ HoodieTableMetaClient dataTableMetaClient =
546+ HoodieTestUtils .createMetaClient (new HadoopStorageConfiguration (HadoopConfigurations .getHadoopConf (conf )), new Path (metadataTableBasePath ).getParent ().getParent ().toString ());
547+ int metadataPartitions = dataTableMetaClient .getTableConfig ().getMetadataPartitions ().size ();
548+ assertThat ("Instants needed to sync to metadata table do not match" , completedTimeline .countInstants (), is (metadataPartitions ));
534549 assertThat (completedTimeline .lastInstant ().get ().getTimestamp (), startsWith (HoodieTableMetadata .SOLO_COMMIT_TIMESTAMP ));
535550
536551 instant = mockWriteWithMetadata ();
537552 metadataTableMetaClient .reloadActiveTimeline ();
538553 completedTimeline = metadataTableMetaClient .getActiveTimeline ().filterCompletedInstants ();
539- assertThat ("One instant need to sync to metadata table" , completedTimeline .countInstants (), is (2 ));
554+ assertThat ("One instant need to sync to metadata table" , completedTimeline .countInstants (), is (metadataPartitions + 1 ));
540555 assertThat (completedTimeline .lastInstant ().get ().getTimestamp (), is (instant ));
541556 }
542557
0 commit comments