message: Do not log an error when `tf.py_func()` raises `StopIteration`.
  This reduces log spam when using `tf.py_func()` to wrap a Python generator,
  such as when using `tf.data.Dataset.from_generator()`.
diff: @@ -176,7 +176,8 @@ string PyExcFetch() {
  }
  // Calls the registered py function through the trampoline.
  -Status DoCallPyFunc(PyCall* call) {
  +Status DoCallPyFunc(PyCall* call, bool* out_log_on_error) {
  + *out_log_on_error = true;
  PyObject* trampoline = GetPyTrampoline();
  if (trampoline == nullptr) {
  return errors::InvalidArgument(
  @@ -196,6 +197,7 @@ Status DoCallPyFunc(PyCall* call) {
  PyErr_ExceptionMatches(PyExc_TypeError)) {
  return errors::InvalidArgument(PyExcFetch());
  } else if (PyErr_ExceptionMatches(PyExc_StopIteration)) {
  + *out_log_on_error = false;
  return errors::OutOfRange(PyExcFetch());
  } else if (PyErr_ExceptionMatches(PyExc_MemoryError)) {
  return errors::ResourceExhausted(PyExcFetch());
  @@ -426,11 +428,19 @@ class PyFuncOp : public OpKernel {
  PyGILState_STATE py_threadstate;
  py_threadstate = PyGILState_Ensure();
  - Status s = DoCallPyFunc(&call);
  + bool log_on_error;
  + Status s = DoCallPyFunc(&call, &log_on_error);
  PyGILState_Release(py_threadstate);
  // Ensures that GIL is released even when !s.ok().
  - OP_REQUIRES_OK(ctx, s);
  + if (!s.ok()) {
  + if (log_on_error) {
  + ctx->CtxFailureWithWarning(s);
  + } else {
  + ctx->CtxFailure(s);
  + }
  + return;
  + }
  OP_REQUIRES(ctx, static_cast<int32>(call.out.size()) == ctx->num_outputs(),
  errors::InvalidArgument(token_, " returns ", call.out.size(),
  