[apache#30941]fix upgrade test due to missed config ConsumerPollingTi…

…meout in KafkaIOTranslation

sdks/java/io/kafka/upgrade/src/main/java/org/apache/beam/sdk/io/kafka/upgrade/KafkaIOTranslation.java

@@ -174,6 +174,9 @@ public Row toConfigRow(Read<?, ?> transform) {
       if (transform.getStopReadTime() != null) {
         fieldValues.put("stop_read_time", transform.getStopReadTime());
       }
+      if (transform.getConsumerPollingTimeout() != null) {
+        fieldValues.put("consumer_polling_timeout", transform.getConsumerPollingTimeout());
+      }
 
       fieldValues.put(
           "is_commit_offset_finalize_enabled", transform.isCommitOffsetsInFinalizeEnabled());
@@ -321,6 +324,12 @@ public Row toConfigRow(Read<?, ?> transform) {
           transform =
               transform.withMaxReadTime(org.joda.time.Duration.millis(maxReadTime.toMillis()));
         }
+        Duration consumerPollingTimeout = configRow.getValue("consumer_polling_timeout");
+        if (consumerPollingTimeout != null) {
+          transform =
+              transform.withConsumerPollingTimeout(
+                  org.joda.time.Duration.millis(consumerPollingTimeout.toMillis()));
+        }
         Instant startReadTime = configRow.getValue("start_read_time");
         if (startReadTime != null) {
           transform = transform.withStartReadTime(startReadTime);