diff --git a/.changeset/search-mele-kalikimaka.md b/.changeset/search-mele-kalikimaka.md new file mode 100644 index 0000000000..4f1faff02f --- /dev/null +++ b/.changeset/search-mele-kalikimaka.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-search-backend-module-elasticsearch': patch +--- + +Fixed a bug that could cause the backstage backend to unexpectedly terminate when client errors were encountered during the indexing process. diff --git a/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngineIndexer.test.ts b/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngineIndexer.test.ts index 92193976b4..94b59fcf42 100644 --- a/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngineIndexer.test.ts +++ b/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngineIndexer.test.ts @@ -250,4 +250,33 @@ describe('ElasticSearchSearchEngineIndexer', () => { // Final deletion shouldn't be called. expect(deleteSpy).not.toHaveBeenCalled(); }); + + it('handles bulk client rejection', async () => { + // Given an ES client wrapper that rejects an error + const expectedError = new Error('HTTP Timeout'); + const mockClientWrapper = ElasticSearchClientWrapper.fromClientOptions({ + node: 'http://localhost:9200', + Connection: mock.getConnection(), + }); + mockClientWrapper.bulk = jest.fn().mockRejectedValue(expectedError); + + // And a search engine indexer that uses that client wrapper + indexer = new ElasticSearchSearchEngineIndexer({ + type: 'some-type', + indexPrefix: '', + indexSeparator: '-index__', + alias: 'some-type-index__search', + logger: getVoidLogger(), + elasticSearchClientWrapper: mockClientWrapper, + batchSize: 1000, + }); + + // When the indexer is run in the test pipeline + const { error } = await TestPipeline.fromIndexer(indexer) + .withDocuments([{ title: 'a', location: 'a', text: '/a' }]) + .execute(); + + // Then the pipeline should have received the expected error + expect(error).toBe(expectedError); + }); }); diff --git a/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngineIndexer.ts b/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngineIndexer.ts index 29f20ea650..3f842976e5 100644 --- a/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngineIndexer.ts +++ b/plugins/search-backend-module-elasticsearch/src/engines/ElasticSearchSearchEngineIndexer.ts @@ -60,6 +60,7 @@ export class ElasticSearchSearchEngineIndexer extends BatchSearchEngineIndexer { private readonly sourceStream: Readable; private readonly elasticSearchClientWrapper: ElasticSearchClientWrapper; private bulkResult: Promise; + private bulkClientError?: Error; constructor(options: ElasticSearchSearchEngineIndexerOptions) { super({ batchSize: options.batchSize }); @@ -94,6 +95,11 @@ export class ElasticSearchSearchEngineIndexer extends BatchSearchEngineIndexer { }, refreshOnCompletion: that.indexName, }); + + // Safely catch errors thrown by the bulk helper client, e.g. HTTP timeouts + this.bulkResult.catch(e => { + this.bulkClientError = e; + }); } async initialize(): Promise { @@ -197,6 +203,11 @@ export class ElasticSearchSearchEngineIndexer extends BatchSearchEngineIndexer { * backpressure in other parts of the indexing pipeline. */ private isReady(): Promise { + // Early exit if the underlying ES client encountered an error. + if (this.bulkClientError) { + return Promise.reject(this.bulkClientError); + } + return new Promise(resolve => { const interval = setInterval(() => { if (this.received === this.processed) {