[FLINK-19156] Migration of transactionIdHint in Kafka is never applied

• Type: Bug
• Status: Open
• Priority: Not a Priority
• Resolution: Unresolved
• Affects Version/s: 1.9.0
• Component/s: Connectors / Kafka
• Labels: auto-deprioritized-critical auto-deprioritized-major auto-deprioritized-minor
• Assignee: Unassigned
• Reporter: Dawid Wysakowicz
• Created: 07/Sep/20 10:21
• Updated: 15/Dec/21 10:39

Description

The code that checks if we should migrate the transaction id is as follows:

@Deprecated
private static final ListStateDescriptor<FlinkKafkaProducer.NextTransactionalIdHint> NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR =
	new ListStateDescriptor<>("next-transactional-id-hint", TypeInformation.of(NextTransactionalIdHint.class));

if (context.getOperatorStateStore().getRegisteredStateNames().contains(NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR)) {
	migrateNextTransactionalIdHindState(context);
}

The condition in if statement is never met because it checks if a Set<String> contains object of type ListStateDescriptor