Fix ncf test for keras (#6355)

* Fix ncf test for keras

* add a todo for batch_size and eval_batch_size for ncf keras

* lint fix

* fix typos

* Lint fix

* fix lint

* resolve pr comment

* resolve pr comment

official/recommendation/data_pipeline.py

@@ -640,6 +640,9 @@ class DummyConstructor(threading.Thread):
   def stop_loop(self):
     pass
 
+  def increment_request_epoch(self):
+    pass
+
   @staticmethod
   def make_input_fn(is_training):
     """Construct training input_fn that uses synthetic data."""

official/recommendation/ncf_keras_main.py

@@ -151,8 +151,21 @@ def _get_keras_model(params):
 
 def run_ncf(_):
   """Run NCF training and eval with Keras."""
+  # TODO(seemuch): Support different train and eval batch sizes
+  if FLAGS.eval_batch_size != FLAGS.batch_size:
+    tf.logging.warning(
+        "The Keras implementation of NCF currently does not support batch_size "
+        "!= eval_batch_size ({} vs. {}). Overriding eval_batch_size to match "
+        "batch_size".format(FLAGS.eval_batch_size, FLAGS.batch_size)
+        )
+    FLAGS.eval_batch_size = FLAGS.batch_size
+
   params = ncf_common.parse_flags(FLAGS)
 
+  # ncf_common rounds eval_batch_size (this is needed due to a reshape during
+  # eval). This carries over that rounding to batch_size as well.
+  params['batch_size'] = params['eval_batch_size']
+
   num_users, num_items, num_train_steps, num_eval_steps, producer = (
       ncf_common.get_inputs(params))
 

official/recommendation/ncf_test.py

@@ -201,7 +201,6 @@ class NcfTest(tf.test.TestCase):
 
   @mock.patch.object(rconst, "SYNTHETIC_BATCHES_PER_EPOCH", 100)
   def test_end_to_end_keras(self):
-    self.skipTest("TODO: fix synthetic data with keras")
     integration.run_synthetic(
         ncf_keras_main.main, tmp_root=self.get_temp_dir(), max_train=None,
         extra_flags=self._BASE_END_TO_END_FLAGS +
@@ -209,7 +208,6 @@ class NcfTest(tf.test.TestCase):
 
   @mock.patch.object(rconst, "SYNTHETIC_BATCHES_PER_EPOCH", 100)
   def test_end_to_end_keras_mlperf(self):
-    self.skipTest("TODO: fix synthetic data with keras")
     integration.run_synthetic(
         ncf_keras_main.main, tmp_root=self.get_temp_dir(), max_train=None,
         extra_flags=self._BASE_END_TO_END_FLAGS +