# Intermittent error while ingesting files in pipeline

**URL:** <https://discuss.elastic.co/t/intermittent-error-while-ingesting-files-in-pipeline/354638>\
**Category:** Elasticsearch\
**Tags:** language-clients, runtime-fields\
**Created:** [March 4, 2024, 12:41pm UTC](https://discuss.elastic.co/t/intermittent-error-while-ingesting-files-in-pipeline/354638 "2024-03-04T12:41:53Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![ChristopheL77](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/christophel77/32/132311_2.png) [@ChristopheL77](https://discuss.elastic.co/u/ChristopheL77)\
**Post date:** [March 4, 2024, 12:41pm UTC](https://discuss.elastic.co/t/intermittent-error-while-ingesting-files-in-pipeline/354638/1 "2024-03-04T12:41:53Z")

</div>

Hi,

When indexing files with BulkOperation with NEST (7.17.5) client and Elastic Search 8.8.0 or 8.12.2, after a few successful operations we have an error message. If we retry, the error will happen on different file.  
If I disable directstreaming on the request, the error doesn't occur anymore.

```auto
BulkDescriptor bulkDescriptorWithPipeline = new BulkDescriptor().Pipeline(EsConstants.FilesPipeline);

// Adding this line seems to fix the issue
.RequestConfiguration(req => req.DisableDirectStreaming());

var props = new Dictionary<string, object>() {
     { "relatedfilejoin", new { parent = parentId, name = "FILE" } },
     { "content", fileAsBase64 },
     { "_fileextension", fileExtension },
     { "_filename", fileName }
};

bulkDescriptorWithPipeline.AddOperation(new BulkIndexOperation<dynamic>(l_props)
{
    Index = indexName,
    Id = fileId,
    Document = props,
    Routing = parentId
});

```

Here is the callstack :

```auto
# FailureReason: Unrecoverable/Unexpected BadRequest while attempting POST on http://127.0.0.1:9200/_bulk?pipeline=filespipeline
 - [1] BadRequest: Node: http://127.0.0.1:9200/ Exception: NotSupportedException Took: 00:00:04.3364656
# Audit exception in step 1 BadRequest:
System.NotSupportedException: The stream does not support concurrent IO read or write operations.
   at System.Net.ConnectStream.InternalWrite(Boolean async, Byte[] buffer, Int32 offset, Int32 size, AsyncCallback callback, Object state)
   at System.Net.ConnectStream.BeginWrite(Byte[] buffer, Int32 offset, Int32 size, AsyncCallback callback, Object state)
   at System.Net.Http.StreamToStreamCopy.TryStartWriteSync(Int32 bytesRead)
   at System.Net.Http.StreamToStreamCopy.StartRead()
--- End of stack trace from previous location where exception was thrown ---
   at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
   at Elasticsearch.Net.HttpConnection.Request[TResponse](RequestData requestData)
   at Elasticsearch.Net.RequestPipeline.CallElasticsearch[TResponse](RequestData requestData)
   at Elasticsearch.Net.Transport`1.Request[TResponse](HttpMethod method, String path, PostData data, IRequestParameters requestParameters)
# Inner Exception: The stream does not support concurrent IO read or write operations.
System.NotSupportedException: The stream does not support concurrent IO read or write operations.
   at System.Net.ConnectStream.InternalWrite(Boolean async, Byte[] buffer, Int32 offset, Int32 size, AsyncCallback callback, Object state)
   at System.Net.ConnectStream.BeginWrite(Byte[] buffer, Int32 offset, Int32 size, AsyncCallback callback, Object state)
   at System.Net.Http.StreamToStreamCopy.TryStartWriteSync(Int32 bytesRead)
   at System.Net.Http.StreamToStreamCopy.StartRead()
--- End of stack trace from previous location where exception was thrown ---
   at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
   at Elasticsearch.Net.HttpConnection.Request[TResponse](RequestData requestData)
   at Elasticsearch.Net.RequestPipeline.CallElasticsearch[TResponse](RequestData requestData)
   at Elasticsearch.Net.Transport`1.Request[TResponse](HttpMethod method, String path, PostData data, IRequestParameters requestParameters)
# Exception:
Elasticsearch.Net.UnexpectedElasticsearchClientException: The stream does not support concurrent IO read or write operations. ---> System.NotSupportedException: The stream does not support concurrent IO read or write operations.
   at System.Net.ConnectStream.InternalWrite(Boolean async, Byte[] buffer, Int32 offset, Int32 size, AsyncCallback callback, Object state)
   at System.Net.ConnectStream.BeginWrite(Byte[] buffer, Int32 offset, Int32 size, AsyncCallback callback, Object state)
   at System.Net.Http.StreamToStreamCopy.TryStartWriteSync(Int32 bytesRead)
   at System.Net.Http.StreamToStreamCopy.StartRead()
--- End of stack trace from previous location where exception was thrown ---
   at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
   at Elasticsearch.Net.HttpConnection.Request[TResponse](RequestData requestData)
   at Elasticsearch.Net.RequestPipeline.CallElasticsearch[TResponse](RequestData requestData)
   at Elasticsearch.Net.Transport`1.Request[TResponse](HttpMethod method, String path, PostData data, IRequestParameters requestParameters)
   --- End of inner exception stack trace ---
   at Elasticsearch.Net.Transport`1.Request[TResponse](HttpMethod method, String path, PostData data, IRequestParameters requestParameters)
   at Nest.ElasticClient.Bulk(IBulkRequest request)

```

I have this warning elasticsearch.log, but I don't know if it is related to the issue above :

```auto
[2024-03-04T13:06:01,583][WARN][o.e.h.AbstractHttpServerTransport] [APT02-7FHMQ93] caught exception while handling client http traffic, closing connection Netty4HttpChannel{localAddress=/127.0.0.1:9200, remoteAddress=/127.0.0.1:57403}
io.netty.handler.codec.PrematureChannelClosureException: Channel closed while still aggregating message
	at io.netty.handler.codec.MessageAggregator.channelInactive(MessageAggregator.java:436) ~[?:?]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:305) ~[?:?]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281) ~[?:?]
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:274) ~[?:?]
	at io.netty.channel.ChannelInboundHandlerAdapter.channelInactive(ChannelInboundHandlerAdapter.java:81) ~[?:?]
	at io.netty.handler.codec.http.HttpContentDecoder.channelInactive(HttpContentDecoder.java:235) ~[?:?]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:305) ~[?:?]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281) ~[?:?]
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:274) ~[?:?]
	at io.netty.channel.ChannelInboundHandlerAdapter.channelInactive(ChannelInboundHandlerAdapter.java:81) ~[?:?]
	at org.elasticsearch.http.netty4.Netty4HttpHeaderValidator.channelInactive(Netty4HttpHeaderValidator.java:186) ~[?:?]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:305) ~[?:?]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281) ~[?:?]
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:274) ~[?:?]
	at io.netty.handler.codec.ByteToMessageDecoder.channelInputClosed(ByteToMessageDecoder.java:411) ~[?:?]
	at io.netty.handler.codec.ByteToMessageDecoder.channelInactive(ByteToMessageDecoder.java:376) ~[?:?]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:305) ~[?:?]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281) ~[?:?]
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:274) ~[?:?]
	at io.netty.channel.ChannelInboundHandlerAdapter.channelInactive(ChannelInboundHandlerAdapter.java:81) ~[?:?]
	at org.elasticsearch.transport.netty4.Netty4WriteThrottlingHandler.channelInactive(Netty4WriteThrottlingHandler.java:109) ~[?:?]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:303) ~[?:?]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281) ~[?:?]
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:274) ~[?:?]
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelInactive(DefaultChannelPipeline.java:1405) ~[?:?]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:301) ~[?:?]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281) ~[?:?]
	at io.netty.channel.DefaultChannelPipeline.fireChannelInactive(DefaultChannelPipeline.java:901) ~[?:?]
	at io.netty.channel.AbstractChannel$AbstractUnsafe$7.run(AbstractChannel.java:813) ~[?:?]
	at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:174) ~[?:?]
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:167) ~[?:?]
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470) ~[?:?]
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:566) ~[?:?]
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997) ~[?:?]
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[?:?]
	at java.lang.Thread.run(Thread.java:1583) ~[?:?]
[2024-03-04T13:06:08,730][INFO][o.e.i.b.HierarchyCircuitBreakerService] [APT02-7FHMQ93] attempting to trigger G1GC due to high heap usage [1043445192]
[2024-03-04T13:06:08,742][INFO][o.e.i.b.HierarchyCircuitBreakerService] [APT02-7FHMQ93] GC did bring memory usage down, before [1043445192], after [685622688], allocations [6], duration [11]

```

Any idea to solve this issue would be appreciated 🙂

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [April 1, 2024, 12:42pm UTC](https://discuss.elastic.co/t/intermittent-error-while-ingesting-files-in-pipeline/354638/2 "2024-04-01T12:42:45Z")

</div>

Closed on April 1, 2024, 12:42pm UTC

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.