diff --git a/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/stream/BinlogEventListener.java b/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/stream/BinlogEventListener.java index 0de9c954f7..8967cca248 100644 --- a/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/stream/BinlogEventListener.java +++ b/data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/stream/BinlogEventListener.java @@ -248,6 +248,13 @@ void handleTableMapEvent(com.github.shyiko.mysql.binlog.event.Event event) { final String fullTableName = databaseName + SEPARATOR + tableName; if (!isTableOfInterest(fullTableName)) { + // The binlog table id is assigned by the server from its table definition cache and is + // not stable: the same table id can later refer to a different table, for example after + // a server restart or after the table was evicted from and reloaded into the cache. + // Drop any mapping we still hold for this table id. Otherwise row events for this + // table, which is not of interest, would be decoded with the stale metadata of a table + // of interest and produce corrupted records. + removeStaleTableMetadata(eventData.getTableId(), fullTableName); return; } @@ -267,6 +274,15 @@ void handleTableMapEvent(com.github.shyiko.mysql.binlog.event.Event event) { tableMetadataMap.put(eventData.getTableId(), tableMetadata); } + private void removeStaleTableMetadata(final long tableId, final String currentTableName) { + final TableMetadata staleTableMetadata = tableMetadataMap.remove(tableId); + if (staleTableMetadata != null) { + LOG.info("Table id {} now refers to table {}, which is not a table of interest. " + + "Removed the stale metadata of table {} held for this table id.", + tableId, currentTableName, staleTableMetadata.getFullTableName()); + } + } + private Map getSetStrValues(final TableMapEventData eventData) { return getStrValuesMap(eventData, MySQLDataType.SET); } diff --git a/data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/stream/BinlogEventListenerTest.java b/data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/stream/BinlogEventListenerTest.java index 925ca41922..55a4f96b4d 100644 --- a/data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/stream/BinlogEventListenerTest.java +++ b/data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/stream/BinlogEventListenerTest.java @@ -7,6 +7,8 @@ import com.github.shyiko.mysql.binlog.BinaryLogClient; import com.github.shyiko.mysql.binlog.event.EventType; +import com.github.shyiko.mysql.binlog.event.TableMapEventData; +import com.github.shyiko.mysql.binlog.event.TableMapEventMetadata; import com.github.shyiko.mysql.binlog.event.UpdateRowsEventData; import io.micrometer.core.instrument.Counter; import io.micrometer.core.instrument.Metrics; @@ -425,4 +427,125 @@ private void verifyHandlerCallHelper() { Runnable capturedRunnable = runnableArgumentCaptor.getValue(); capturedRunnable.run(); } + + @Test + void test_given_TableMap_event_for_table_not_of_interest_then_removes_stale_table_metadata() throws Exception { + final long tableId = 1234L; + final String databaseName = "test_database"; + final String tableOfInterest = "table_of_interest"; + final String tableNotOfInterest = "table_not_of_interest"; + final String fullTableOfInterest = databaseName + "." + tableOfInterest; + + // Only tableOfInterest is included in the sync set + when(dbTableMetadata.getTableColumnDataTypeMap()) + .thenReturn(Map.of(fullTableOfInterest, Map.of("col1", "int", "col2", "varchar"))); + final BinlogEventListener objectUnderTest = createObjectUnderTestWithMockedExecutors(); + + // The listener is holding a mapping from tableId to the table of interest, which was + // built from an earlier TABLE_MAP event. + final TableMetadata staleTableMetadata = TableMetadata.builder() + .withDatabaseName(databaseName) + .withTableName(tableOfInterest) + .withColumnNames(List.of("col1", "col2")) + .withPrimaryKeys(List.of("col1")) + .build(); + setTableMetadataMap(objectUnderTest, tableId, staleTableMetadata); + assertThat(objectUnderTest.isValidTableId(tableId), is(true)); + + // MySQL reassigns the same tableId to a table that is not part of the sync set. + objectUnderTest.handleTableMapEvent(createTableMapEvent(tableId, databaseName, tableNotOfInterest)); + + // The stale mapping must be gone, otherwise row events of tableNotOfInterest would be + // decoded with the schema of tableOfInterest. + assertThat(getTableMetadataMap(objectUnderTest).containsKey(tableId), is(false)); + assertThat(objectUnderTest.isValidTableId(tableId), is(false)); + } + + @Test + void test_given_TableMap_event_for_table_not_of_interest_and_no_existing_mapping_then_no_mapping_is_added() throws Exception { + final long tableId = 1234L; + final String databaseName = "test_database"; + final String fullTableOfInterest = databaseName + ".table_of_interest"; + + when(dbTableMetadata.getTableColumnDataTypeMap()) + .thenReturn(Map.of(fullTableOfInterest, Map.of("col1", "int"))); + final BinlogEventListener objectUnderTest = createObjectUnderTestWithMockedExecutors(); + + objectUnderTest.handleTableMapEvent(createTableMapEvent(tableId, databaseName, "table_not_of_interest")); + + assertThat(getTableMetadataMap(objectUnderTest).isEmpty(), is(true)); + assertThat(objectUnderTest.isValidTableId(tableId), is(false)); + } + + @Test + void test_given_TableMap_event_for_table_of_interest_then_overwrites_existing_mapping() throws Exception { + final long tableId = 1234L; + final String databaseName = "test_database"; + final String previousTable = "previous_table_of_interest"; + final String currentTable = "current_table_of_interest"; + + when(dbTableMetadata.getTableColumnDataTypeMap()).thenReturn(Map.of( + databaseName + "." + previousTable, Map.of("col1", "int"), + databaseName + "." + currentTable, Map.of("col1", "int", "col2", "varchar"))); + final BinlogEventListener objectUnderTest = createObjectUnderTestWithMockedExecutors(); + + final TableMetadata staleTableMetadata = TableMetadata.builder() + .withDatabaseName(databaseName) + .withTableName(previousTable) + .withColumnNames(List.of("col1")) + .withPrimaryKeys(List.of("col1")) + .build(); + setTableMetadataMap(objectUnderTest, tableId, staleTableMetadata); + + final com.github.shyiko.mysql.binlog.event.Event tableMapEvent = + createTableMapEvent(tableId, databaseName, currentTable); + final TableMapEventData tableMapEventData = tableMapEvent.getData(); + final TableMapEventMetadata tableMapEventMetadata = mock(TableMapEventMetadata.class); + when(tableMapEventData.getEventMetadata()).thenReturn(tableMapEventMetadata); + when(tableMapEventMetadata.getColumnNames()).thenReturn(List.of("col1", "col2")); + when(tableMapEventMetadata.getSimplePrimaryKeys()).thenReturn(List.of(0)); + + objectUnderTest.handleTableMapEvent(tableMapEvent); + + final TableMetadata updatedTableMetadata = getTableMetadataMap(objectUnderTest).get(tableId); + assertThat(updatedTableMetadata.getTableName(), is(currentTable)); + assertThat(updatedTableMetadata.getColumnNames(), is(List.of("col1", "col2"))); + assertThat(objectUnderTest.isValidTableId(tableId), is(true)); + } + + private com.github.shyiko.mysql.binlog.event.Event createTableMapEvent( + final long tableId, final String databaseName, final String tableName) { + final com.github.shyiko.mysql.binlog.event.Event tableMapEvent = + mock(com.github.shyiko.mysql.binlog.event.Event.class); + final TableMapEventData tableMapEventData = mock(TableMapEventData.class); + when(tableMapEvent.getData()).thenReturn(tableMapEventData); + lenient().when(tableMapEventData.getTableId()).thenReturn(tableId); + when(tableMapEventData.getDatabase()).thenReturn(databaseName); + when(tableMapEventData.getTable()).thenReturn(tableName); + return tableMapEvent; + } + + private BinlogEventListener createObjectUnderTestWithMockedExecutors() { + try (final MockedStatic executorsMockedStatic = mockStatic(Executors.class)) { + executorsMockedStatic.when(() -> Executors.newFixedThreadPool(anyInt(), any(ThreadFactory.class))) + .thenReturn(eventListnerExecutorService); + executorsMockedStatic.when(Executors::newSingleThreadExecutor).thenReturn(checkpointManagerExecutorService); + executorsMockedStatic.when(Executors::defaultThreadFactory).thenReturn(threadFactory); + return createObjectUnderTest(); + } + } + + private void setTableMetadataMap(final BinlogEventListener binlogEventListener, + final long tableId, + final TableMetadata tableMetadata) throws Exception { + getTableMetadataMap(binlogEventListener).put(tableId, tableMetadata); + } + + @SuppressWarnings("unchecked") + private Map getTableMetadataMap(final BinlogEventListener binlogEventListener) + throws Exception { + final Field tableMetadataMapField = BinlogEventListener.class.getDeclaredField("tableMetadataMap"); + tableMetadataMapField.setAccessible(true); + return (Map) tableMetadataMapField.get(binlogEventListener); + } }