Make OutboxPollingAutomaticIndexingDynamicShardingRebalancingIT more …

…robust

Especially on Windows

integrationtest/mapper/orm-coordination-outbox-polling/src/test/java/org/hibernate/search/integrationtest/mapper/orm/coordination/outboxpolling/automaticindexing/OutboxPollingAutomaticIndexingDynamicShardingRebalancingIT.java

@@ -23,6 +23,7 @@
 import org.hibernate.cfg.Environment;
 import org.hibernate.search.engine.backend.analysis.AnalyzerNames;
 import org.hibernate.search.integrationtest.mapper.orm.coordination.outboxpolling.testsupport.util.OutboxAgentDisconnectionSimulator;
+import org.hibernate.search.integrationtest.mapper.orm.coordination.outboxpolling.testsupport.util.OutboxEventFilter;
 import org.hibernate.search.integrationtest.mapper.orm.coordination.outboxpolling.testsupport.util.PerSessionFactoryIndexingCountHelper;
 import org.hibernate.search.integrationtest.mapper.orm.coordination.outboxpolling.testsupport.util.TestingOutboxPollingInternalConfigurer;
 import org.hibernate.search.mapper.orm.coordination.outboxpolling.cfg.impl.HibernateOrmMapperOutboxPollingImplSettings;
@@ -68,6 +69,8 @@ public class OutboxPollingAutomaticIndexingDynamicShardingRebalancingIT {
 	private final List<OutboxAgentDisconnectionSimulator> outboxAgentDisconnectionSimulators =
 			new ArrayList<>();
 
+	private final OutboxEventFilter eventFilter = new OutboxEventFilter();
+
 	public void setup() {
 		setup( "create-drop" );
 		setup( "none" );
@@ -92,6 +95,7 @@ private void setup(String hbm2ddlAction) {
 				.with( indexingCountHelper::bind )
 				.withProperty( HibernateOrmMapperOutboxPollingImplSettings.COORDINATION_INTERNAL_CONFIGURER,
 						new TestingOutboxPollingInternalConfigurer()
+								.outboxEventFilter( eventFilter )
 								.agentDisconnectionSimulator( outboxAgentDisconnectionSimulator ) )
 				.withProperty( "hibernate.search.coordination.event_processor.polling_interval", POLLING_INTERVAL )
 				.withProperty( "hibernate.search.coordination.event_processor.pulse_expiration", PULSE_EXPIRATION )
@@ -120,6 +124,10 @@ public void agentLeft() {
 			}
 		} );
 
+		// The filter is there to make sure we don't consume all events while we're creating them,
+		// which apparently can happen on Windows.
+		eventFilter.showAllEvents();
+
 		// Stop the last factory as soon as it's processed at least one entity
 		await()
 				.pollInterval( 1, TimeUnit.MILLISECONDS )
@@ -169,6 +177,10 @@ public void agentExpired() {
 			}
 		} );
 
+		// The filter is there to make sure we don't consume all events while we're creating them,
+		// which apparently can happen on Windows.
+		eventFilter.showAllEvents();
+
 		// Prevent the last factory from accessing the database as soon as it's processed at least one entity,
 		// so that its registration ultimately expires
 		await()
@@ -212,13 +224,16 @@ public void agentJoined() {
 			}
 		} );
 
+		// The filter is there to make sure we don't consume all events while we're creating them,
+		// which apparently can happen on Windows.
+		eventFilter.showAllEvents();
+
 		// Start a new factory as soon as all others have processed at least one entity
 		await()
 				.pollInterval( 1, TimeUnit.MILLISECONDS )
 				.untilAsserted( () -> indexingCountHelper.indexingCounts().assertForEachSessionFactory()
 						.allSatisfy( c -> assertThat( c ).isNotZero() ) );
 		setup( "none" );
-		int newShardCount = initialShardCount + 1;
 
 		backendMock.verifyExpectationsMet();
 		// All works must be executed exactly once