docs: private List<KafkaTopicPartitionState<KPH>> createPartitionStateHolders(
  			Map<KafkaTopicPartition, Long> partitionsToInitialOffsets,
  			int timestampWatermarkMode,
  			SerializedValue<AssignerWithPeriodicWatermarks<T>> watermarksPeriodic,
  			SerializedValue<AssignerWithPunctuatedWatermarks<T>> watermarksPunctuated,
  			ClassLoader userCodeClassLoader) throws IOException, ClassNotFoundException {
  
  		// CopyOnWrite as adding discovered partitions could happen in parallel
  		// while different threads iterate the partitions list
  		List<KafkaTopicPartitionState<KPH>> partitionStates = new CopyOnWriteArrayList<>();
  
  		switch (timestampWatermarkMode) {
  			case NO_TIMESTAMPS_WATERMARKS: {
  				for (Map.Entry<KafkaTopicPartition, Long> partitionEntry : partitionsToInitialOffsets.entrySet()) {
  					// create the kafka version specific partition handle
  					KPH kafkaHandle = createKafkaPartitionHandle(partitionEntry.getKey());
  
  					KafkaTopicPartitionState<KPH> partitionState =
  							new KafkaTopicPartitionState<>(partitionEntry.getKey(), kafkaHandle);
  					partitionState.setOffset(partitionEntry.getValue());
  
  					partitionStates.add(partitionState);
  				}
  
  				return partitionStates;
  			}
  
  			case PERIODIC_WATERMARKS: {
  				for (Map.Entry<KafkaTopicPartition, Long> partitionEntry : partitionsToInitialOffsets.entrySet()) {
  					KPH kafkaHandle = createKafkaPartitionHandle(partitionEntry.getKey());
  
  					AssignerWithPeriodicWatermarks<T> assignerInstance =
  							watermarksPeriodic.deserializeValue(userCodeClassLoader);
  
  					KafkaTopicPartitionStateWithPeriodicWatermarks<T, KPH> partitionState =
  							new KafkaTopicPartitionStateWithPeriodicWatermarks<>(
  									partitionEntry.getKey(),
  									kafkaHandle,
  									assignerInstance);
  
  					partitionState.setOffset(partitionEntry.getValue());
  
  					partitionStates.add(partitionState);
  				}
  
  				return partitionStates;
  			}
  
  			case PUNCTUATED_WATERMARKS: {
  				for (Map.Entry<KafkaTopicPartition, Long> partitionEntry : partitionsToInitialOffsets.entrySet()) {
  					KPH kafkaHandle = createKafkaPartitionHandle(partitionEntry.getKey());
  
  					AssignerWithPunctuatedWatermarks<T> assignerInstance =
  							watermarksPunctuated.deserializeValue(userCodeClassLoader);
  
  					KafkaTopicPartitionStateWithPunctuatedWatermarks<T, KPH> partitionState =
  							new KafkaTopicPartitionStateWithPunctuatedWatermarks<>(
  									partitionEntry.getKey(),
  									kafkaHandle,
  									assignerInstance);
  
  					partitionState.setOffset(partitionEntry.getValue());
  
  					partitionStates.add(partitionState);
  				}
  
  				return partitionStates;
  			}
  			default:
  				// cannot happen, add this as a guard for the future
  				throw new RuntimeException();
  		}
  	}